feat: Add basic authentication tokens and make them available to users

Basic authentication creates tokens which are longer lived in order to e.g. use them for pipelines or scripts. The tokens can be created and deleted in by the user. In most places authentication with oauth is made possible now.

backend/capellacollab/alembic/versions/c9f30ccd4650_add_basic_auth_token.py

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Add basic auth token
+
+Revision ID: c9f30ccd4650
+Revises: d8cf851562cd
+Create Date: 2023-09-06 14:42:53.016924
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c9f30ccd4650"
+down_revision = "d8cf851562cd"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "basic_auth_token",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("hash", sa.String(), nullable=False),
+        sa.Column("expiration_date", sa.Date(), nullable=False),
+        sa.Column("description", sa.String(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_basic_auth_token_id"),
+        "basic_auth_token",
+        ["id"],
+        unique=False,
+    )
+    # ### end Alembic commands ###

backend/capellacollab/core/authentication/basic_auth.py

@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+import logging
+
+import fastapi
+from fastapi import security, status
+
+from capellacollab.core import database
+from capellacollab.users import crud as user_crud
+from capellacollab.users.tokens import crud as token_crud
+
+logger = logging.getLogger(__name__)
+
+
+class HTTPBasicAuth(security.HTTPBasic):
+    async def __call__(  # type: ignore
+        self, request: fastapi.Request
+    ) -> security.HTTPBasicCredentials | None:
+        credentials: security.HTTPBasicCredentials | None = (
+            await super().__call__(request)
+        )
+        if not credentials:
+            if self.auto_error:
+                raise fastapi.HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="Not authenticated",
+                    headers={"WWW-Authenticate": "Basic"},
+                )
+            return None
+        with database.SessionLocal() as session:
+            user = user_crud.get_user_by_name(session, credentials.username)
+            token_data = None
+            if user:
+                token_data = token_crud.get_token(
+                    session, credentials.password, user.id
+                )
+            if not token_data or not user or token_data.user_id != user.id:
+                logger.error("Token invalid for user %s", credentials.username)
+                if self.auto_error:
+                    raise fastapi.HTTPException(
+                        status_code=status.HTTP_401_UNAUTHORIZED,
+                        detail={
+                            "err_code": "TOKEN_INVALID",
+                            "reason": "The used token is not valid.",
+                        },
+                        headers={"WWW-Authenticate": "Basic"},
+                    )
+                return None
+
+            if token_data.expiration_date < datetime.date.today():
+                logger.error("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": "Basic"},
+                    )
+                return None
+        return credentials

backend/capellacollab/core/authentication/helper.py

@@ -7,5 +7,7 @@
 from capellacollab.config import config
 
 
-def get_username(token: dict[str, t.Any]) -> str:
-    return token[config["authentication"]["jwt"]["usernameClaim"]].strip()
+def get_username(token: dict[str, t.Any], is_bearer: bool = True) -> str:
+    if is_bearer:
+        return token[config["authentication"]["jwt"]["usernameClaim"]].strip()
+    return token["username"]

backend/capellacollab/core/authentication/injectables.py

@@ -3,11 +3,15 @@
 
 from __future__ import annotations
 
+import logging
+import typing as t
+
 import fastapi
 from fastapi import status
+from fastapi.security import utils as fastapi_utils
 
 from capellacollab.core import database
-from capellacollab.core.authentication import jwt_bearer
+from capellacollab.core.authentication import basic_auth, jwt_bearer
 from capellacollab.projects import crud as projects_crud
 from capellacollab.projects import models as projects_models
 from capellacollab.projects.users import crud as projects_users_crud
@@ -17,6 +21,42 @@
 
 from . import helper
 
+logger = logging.getLogger(__name__)
+
+
+async def get_username(request: fastapi.Request) -> str:
+    token, scheme = await get_token(request)
+
+    if not token:
+        raise fastapi.HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token is none and username cannot be derived",
+        )
+
+    return helper.get_username(token, is_bearer=scheme.lower() == "bearer")
+
+
+async def get_token(
+    request: fastapi.Request, auto_error: bool = True
+) -> tuple[dict[str, t.Any] | None, str]:
+    scheme, _ = fastapi_utils.get_authorization_scheme_param(
+        request.headers.get("Authorization")
+    )
+    token = None
+    match scheme.lower():
+        case "bearer":
+            token = await jwt_bearer.JWTBearer(auto_error=auto_error)(request)
+        case "basic":
+            basic_creds = await basic_auth.HTTPBasicAuth(
+                auto_error=auto_error
+            )(request)
+            if basic_creds:
+                token = basic_creds.model_dump()
+        case _:
+            logger.error("Authentification scheme unknown")
+
+    return token, scheme
+
 
 class RoleVerification:
     def __init__(self, required_role: users_models.Role, verify: bool = True):
@@ -25,10 +65,9 @@ def __init__(self, required_role: users_models.Role, verify: bool = True):
 
     def __call__(
         self,
-        token=fastapi.Depends(jwt_bearer.JWTBearer()),
+        username=fastapi.Depends(get_username),
         db=fastapi.Depends(database.get_db),
     ) -> bool:
-        username = helper.get_username(token)
         if not (user := users_crud.get_user_by_name(db, username)):
             if self.verify:
                 raise fastapi.HTTPException(
@@ -73,10 +112,9 @@ def __init__(
     def __call__(
         self,
         project_slug: str,
-        token=fastapi.Depends(jwt_bearer.JWTBearer()),
+        username=fastapi.Depends(get_username),
         db=fastapi.Depends(database.get_db),
     ) -> bool:
-        username = helper.get_username(token)
         if not (user := users_crud.get_user_by_name(db, username)):
             if self.verify:
                 raise fastapi.HTTPException(

backend/capellacollab/core/authentication/provider/azure/routes.py

@@ -94,9 +94,11 @@ async def logout(jwt_decoded=fastapi.Depends(JWTBearer())):
 @router.get("/tokens", name="Validate the token")
 async def validate_token(
     scope: Role | None,
-    token=fastapi.Depends(JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
     db: orm.Session = fastapi.Depends(database.get_db),
 ):
     if scope and scope.ADMIN:
-        auth_injectables.RoleVerification(required_role=Role.ADMIN)(token, db)
-    return token
+        auth_injectables.RoleVerification(required_role=Role.ADMIN)(
+            username, db
+        )
+    return username

backend/capellacollab/core/authentication/provider/oauth/routes.py

@@ -52,9 +52,11 @@ async def logout():
 @router.get("/tokens", name="Validate the token")
 async def validate_token(
     scope: Role | None,
-    token=fastapi.Depends(JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
     db: orm.Session = fastapi.Depends(database.get_db),
 ):
     if scope and scope.ADMIN:
-        auth_injectables.RoleVerification(required_role=Role.ADMIN)(token, db)
-    return token
+        auth_injectables.RoleVerification(required_role=Role.ADMIN)(
+            username, db
+        )
+    return username

backend/capellacollab/core/database/models.py

@@ -21,3 +21,4 @@
 import capellacollab.tools.models
 import capellacollab.users.events.models
 import capellacollab.users.models
+import capellacollab.users.tokens

backend/capellacollab/core/logging/__init__.py

@@ -15,7 +15,7 @@
 
 from capellacollab import config
 from capellacollab.core.authentication import helper as auth_helper
-from capellacollab.core.authentication import jwt_bearer
+from capellacollab.core.authentication import injectables as auth_injectables
 
 LOGGING_LEVEL = config.config["logging"]["level"]
 
@@ -59,8 +59,13 @@ async def dispatch(
         self, request: fastapi.Request, call_next: base.RequestResponseEndpoint
     ):
         username = "anonymous"
-        if token := await jwt_bearer.JWTBearer(auto_error=False)(request):
-            username = auth_helper.get_username(token)
+        token, scheme = await auth_injectables.get_token(
+            request, auto_error=False
+        )
+        if token:
+            username = auth_helper.get_username(
+                token, scheme.lower() == "bearer"
+            )
 
         request.state.user_name = username
 

backend/capellacollab/projects/routes.py

@@ -13,7 +13,6 @@
 from capellacollab.core import database
 from capellacollab.core import logging as core_logging
 from capellacollab.core.authentication import injectables as auth_injectables
-from capellacollab.core.authentication import jwt_bearer
 from capellacollab.projects import injectables as projects_injectables
 from capellacollab.projects.events import routes as projects_events_routes
 from capellacollab.projects.toolmodels import routes as toolmodels_routes
@@ -45,14 +44,14 @@ def get_projects(
         users_injectables.get_own_user
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
     log: logging.LoggerAdapter = fastapi.Depends(
         core_logging.get_request_logger
     ),
 ) -> abc.Sequence[models.DatabaseProject]:
     if auth_injectables.RoleVerification(
         required_role=users_models.Role.ADMIN, verify=False
-    )(token, db):
+    )(username, db):
         log.debug("Fetching all projects")
         return crud.get_projects(db)
 

backend/capellacollab/projects/toolmodels/backups/routes.py

@@ -11,9 +11,7 @@
 
 import capellacollab.settings.modelsources.t4c.repositories.interface as t4c_repository_interface
 from capellacollab.core import credentials, database
-from capellacollab.core.authentication import helper as auth_helper
 from capellacollab.core.authentication import injectables as auth_injectables
-from capellacollab.core.authentication import jwt_bearer
 from capellacollab.projects.toolmodels import (
     injectables as toolmodels_injectables,
 )
@@ -73,7 +71,7 @@ def create_backup(
         toolmodels_injectables.get_existing_capella_model
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
 ):
     git_model = git_injectables.get_existing_git_model(
         body.git_model_id, capella_model, db
@@ -123,7 +121,7 @@ def create_backup(
             k8s_cronjob_id=reference,
             git_model=git_model,
             t4c_model=t4c_model,
-            created_by=auth_helper.get_username(token),
+            created_by=username,
             model=capella_model,
             t4c_username=username,
             t4c_password=password,
@@ -139,7 +137,7 @@ def delete_pipeline(
         injectables.get_existing_pipeline
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
     force: bool = False,
 ):
     try:
@@ -159,7 +157,7 @@ def delete_pipeline(
             force
             and auth_injectables.RoleVerification(
                 required_role=users_models.Role.ADMIN, verify=False
-            )(token=token, db=db)
+            )(username=username, db=db)
         ):
             raise exceptions.PipelineOperationFailedT4CServerUnreachable(
                 exceptions.PipelineOperation.DELETE

backend/capellacollab/projects/users/routes.py

@@ -8,7 +8,6 @@
 
 from capellacollab.core import database
 from capellacollab.core.authentication import injectables as auth_injectables
-from capellacollab.core.authentication import jwt_bearer
 from capellacollab.projects import injectables as projects_injectables
 from capellacollab.projects import models as projects_models
 from capellacollab.users import crud as users_crud
@@ -80,11 +79,11 @@ def get_current_user(
         projects_injectables.get_existing_project
     ),
     db: orm.Session = fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
 ) -> models.ProjectUserAssociation | models.ProjectUser:
     if auth_injectables.RoleVerification(
         required_role=users_models.Role.ADMIN, verify=False
-    )(token, db):
+    )(username, db):
         return models.ProjectUser(
             role=models.ProjectUserRole.ADMIN,
             permission=models.ProjectUserPermission.WRITE,

backend/capellacollab/sessions/hooks/t4c.py

@@ -81,14 +81,15 @@ def configuration_hook(
 
         for repository in t4c_repositories:
             try:
+                token_username = auth_injectables.get_username(token)
                 repo_interface.add_user_to_repository(
                     repository.instance,
                     repository.name,
                     username=user.name,
                     password=environment["T4C_PASSWORD"],
                     is_admin=auth_injectables.RoleVerification(
                         required_role=users_models.Role.ADMIN, verify=False
-                    )(token, db),
+                    )(token_username, db),
                 )
             except requests.RequestException:
                 warnings.append(