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,70 @@
+# 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
+    ) -> tuple[str | None, fastapi.HTTPException | None]:
+        credentials: security.HTTPBasicCredentials | None = (
+            await super().__call__(request)
+        )
+        if not credentials:
+            error = fastapi.HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Not authenticated",
+                headers={"WWW-Authenticate": "Bearer, Basic"},
+            )
+            if self.auto_error:
+                raise error
+            return None, error
+        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)
+                error = fastapi.HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail={
+                        "err_code": "TOKEN_INVALID",
+                        "reason": "The used token is not valid.",
+                    },
+                    headers={"WWW-Authenticate": "Bearer, Basic"},
+                )
+                if self.auto_error:
+                    raise error
+                return None, error
+
+            if token_data.expiration_date < datetime.date.today():
+                logger.error("Token expired for user %s", credentials.username)
+                error = 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"},
+                )
+                if self.auto_error:
+                    raise error
+                return None, error
+        return self.get_username(credentials), None
+
+    def get_username(self, credentials: security.HTTPBasicCredentials) -> str:
+        return credentials.model_dump()["username"]

backend/capellacollab/core/authentication/injectables.py

@@ -3,19 +3,56 @@
 
 from __future__ import annotations
 
+import logging
+
 import fastapi
 from fastapi import status
 
 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
 from capellacollab.projects.users import models as projects_users_models
 from capellacollab.users import crud as users_crud
 from capellacollab.users import models as users_models
 
-from . import helper
+logger = logging.getLogger(__name__)
+
+
+async def get_username(
+    basic: tuple[str | None, fastapi.HTTPException | None] = fastapi.Depends(
+        basic_auth.HTTPBasicAuth(auto_error=False)
+    ),
+    jwt: tuple[str | None, fastapi.HTTPException | None] = fastapi.Depends(
+        jwt_bearer.JWTBearer(auto_error=False)
+    ),
+) -> str:
+    jwt_username, jwt_error = jwt
+    if jwt_username:
+        return jwt_username
+
+    basic_username, basic_error = basic
+    if basic_username:
+        return basic_username
+
+    if jwt_error:
+        raise jwt_error
+    if basic_error:
+        raise basic_error
+
+    raise fastapi.HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Token is none and username cannot be derived",
+        headers={"WWW-Authenticate": "Bearer, Basic"},
+    )
+
+
+async def get_username_not_injectable(request: fastapi.Request):
+    basic = await basic_auth.HTTPBasicAuth(auto_error=False)(request)
+    jwt = await jwt_bearer.JWTBearer(auto_error=False)(request)
+
+    return await get_username(basic=basic, jwt=jwt)
 
 
 class RoleVerification:
@@ -25,10 +62,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 +109,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/jwt_bearer.py

@@ -11,8 +11,8 @@
 
 import capellacollab.users.crud as users_crud
 import capellacollab.users.events.crud as events_crud
+from capellacollab.config import config
 from capellacollab.core import database
-from capellacollab.core.authentication import helper as auth_helper
 
 from . import get_authentication_entrypoint
 
@@ -25,29 +25,42 @@ class JWTBearer(security.HTTPBearer):
     def __init__(self, auto_error: bool = True):
         super().__init__(auto_error=auto_error)
 
-    async def __call__(
+    async def __call__(  # type: ignore
         self, request: fastapi.Request
-    ) -> dict[str, t.Any] | None:
+    ) -> tuple[str | None, fastapi.HTTPException | None]:
         credentials: security.HTTPAuthorizationCredentials | None = (
             await super().__call__(request)
         )
 
         if not credentials or credentials.scheme != "Bearer":
+            error = fastapi.HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Not authenticated",
+                headers={"WWW-Authenticate": "Bearer, Basic"},
+            )
             if self.auto_error:
-                raise fastapi.HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="Not authenticated",
-                    headers={"WWW-Authenticate": "Bearer"},
-                )
-            return None
+                raise error
+            return None, error
         if token_decoded := self.validate_token(credentials.credentials):
             self.initialize_user(token_decoded)
-            return token_decoded
-        return None
+            return self.get_username(token_decoded), None
+        error = fastapi.HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Not authenticated",
+            headers={"WWW-Authenticate": "Bearer, Basic"},
+        )
+        if self.auto_error:
+            raise error
+        return None, error
+
+    def get_username(self, token_decoded: dict[str, str]) -> str:
+        return token_decoded[
+            config["authentication"]["jwt"]["usernameClaim"]
+        ].strip()
 
     def initialize_user(self, token_decoded: dict[str, str]):
         with database.SessionLocal() as session:
-            username: str = auth_helper.get_username(token_decoded)
+            username: str = self.get_username(token_decoded)
             if not users_crud.get_user_by_name(session, username):
                 created_user = users_crud.create_user(session, username)
                 users_crud.update_last_login(session, created_user)

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

@@ -14,7 +14,6 @@
 from capellacollab.config import config
 from capellacollab.core import database
 from capellacollab.core.authentication import injectables as auth_injectables
-from capellacollab.core.authentication.helper import get_username
 from capellacollab.core.authentication.jwt_bearer import JWTBearer
 from capellacollab.core.authentication.schemas import (
     RefreshTokenRequest,
@@ -62,7 +61,9 @@ async def api_get_token(
     )
     access_token = token["id_token"]
 
-    username = get_username(JWTBearer().validate_token(access_token))
+    username = JWTBearer().get_username(
+        JWTBearer().validate_token(access_token)
+    )
 
     if user := users_crud.get_user_by_name(db, username):
         users_crud.update_last_login(db, user)
@@ -94,9 +95,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

@@ -7,7 +7,6 @@
 import capellacollab.users.crud as users_crud
 from capellacollab.core import database
 from capellacollab.core.authentication import injectables as auth_injectables
-from capellacollab.core.authentication.helper import get_username
 from capellacollab.core.authentication.jwt_bearer import JWTBearer
 from capellacollab.core.authentication.schemas import (
     RefreshTokenRequest,
@@ -31,7 +30,9 @@ async def api_get_token(
 ):
     token = get_token(body.code)
 
-    username = get_username(JWTBearer().validate_token(token["access_token"]))
+    username = JWTBearer().get_username(
+        JWTBearer().validate_token(token["access_token"])
+    )
 
     if user := users_crud.get_user_by_name(db, username):
         users_crud.update_last_login(db, user)
@@ -52,9 +53,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

@@ -14,8 +14,7 @@
 from starlette.middleware import base
 
 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"]
 
@@ -58,9 +57,12 @@ class AttachUserNameMiddleware(base.BaseHTTPMiddleware):
     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)
+        try:
+            username = await auth_injectables.get_username_not_injectable(
+                request
+            )
+        except fastapi.HTTPException:
+            username = "anonymous"
 
         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)