WIP: porting basic auth tokens

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: 4c58f4db4f54
+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 = "4c58f4db4f54"
+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,56 @@
+# 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 = 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)
+                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"},
+                )
+            if token_data.expiration_date < datetime.date.today():
+                logger.error("Token expired for user %s", credentials.username)
+                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.",
+                    },
+                )
+        return credentials

backend/capellacollab/core/authentication/helper.py

@@ -7,5 +7,8 @@
 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()
+    else:
+        return token["username"]

backend/capellacollab/core/authentication/injectables.py

@@ -3,11 +3,14 @@
 
 from __future__ import annotations
 
+import logging
+
 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 +20,31 @@
 
 from . import helper
 
+logger = logging.getLogger(__name__)
+
+
+async def get_username(request: fastapi.Request) -> str:
+    scheme, _ = fastapi_utils.get_authorization_scheme_param(
+        request.headers.get("Authorization")
+    )
+    token = None
+    match scheme.lower():
+        case "bearer":
+            token = await jwt_bearer.JWTBearer()(request)
+        case "basic":
+            basic_creds = await basic_auth.HTTPBasicAuth()(request)
+            if basic_creds:
+                token = basic_creds.model_dump()
+        case _:
+            logger.error("Authentification scheme unknown")
+    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")
+
 
 class RoleVerification:
     def __init__(self, required_role: users_models.Role, verify: bool = True):
@@ -25,10 +53,11 @@ 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 type(username) == dict:  # pylint: disable=unidiomatic-typecheck
+            username = helper.get_username(username)
         if not (user := users_crud.get_user_by_name(db, username)):
             if self.verify:
                 raise fastapi.HTTPException(
@@ -73,10 +102,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/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/users/injectables.py

@@ -7,17 +7,15 @@
 from sqlalchemy import orm
 
 from capellacollab.core import database
-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
 
 from . import crud, exceptions, models
 
 
 def get_own_user(
     db: orm.Session = fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
 ) -> models.DatabaseUser:
-    username = auth_helper.get_username(token)
 
     if user := crud.get_user_by_name(db, username):
         return user
@@ -28,10 +26,10 @@ def get_own_user(
 def get_existing_user(
     user_id: int | t.Literal["current"],
     db=fastapi.Depends(database.get_db),
-    token=fastapi.Depends(jwt_bearer.JWTBearer()),
+    username=fastapi.Depends(auth_injectables.get_username),
 ) -> models.DatabaseUser:
     if user_id == "current":
-        return get_own_user(db, token)
+        return get_own_user(db, username)
 
     if user := crud.get_user_by_id(db, user_id):
         return user

backend/capellacollab/users/routes.py

@@ -13,6 +13,7 @@
 from capellacollab.users.events import crud as events_crud
 from capellacollab.users.events import models as events_models
 from capellacollab.users.events import routes as events_routes
+from capellacollab.users.tokens import routes as tokens_routes
 
 from . import crud, injectables, models
 
@@ -135,3 +136,4 @@ def delete_user(
 
 router.include_router(session_routes.users_router, tags=["Users - Sessions"])
 router.include_router(events_routes.router, tags=["Users - History"])
+router.include_router(tokens_routes.router, tags=["Users - Token"])

backend/capellacollab/users/tokens/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0

backend/capellacollab/users/tokens/crud.py

@@ -0,0 +1,66 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+from collections import abc
+
+import argon2
+import sqlalchemy as sa
+from sqlalchemy import orm
+
+from capellacollab.core import credentials
+
+from . import models
+
+
+def create_token(
+    db: orm.Session, user_id: int, description: str
+) -> tuple[models.DatabaseUserTokenModel, str]:
+    password = credentials.generate_password(32)
+    ph = argon2.PasswordHasher()
+    token_data = models.DatabaseUserTokenModel(
+        user_id=user_id,
+        hash=ph.hash(password),
+        expiration_date=datetime.datetime.now() + datetime.timedelta(days=30),
+        description=description,
+    )
+    db.add(token_data)
+    db.commit()
+    return token_data, password
+
+
+def get_token(
+    db: orm.Session, password: str, user_id: int
+) -> models.DatabaseUserTokenModel | None:
+    ph = argon2.PasswordHasher()
+    token_list = get_token_by_user(db, user_id)
+    if token_list:
+        for token in token_list:
+            try:
+                ph.verify(token.hash, password)
+                return token
+            except argon2.exceptions.VerifyMismatchError:
+                pass
+    return None
+
+
+def get_token_by_user(
+    db: orm.Session, user_id: int
+) -> abc.Sequence[models.DatabaseUserTokenModel] | None:
+    return (
+        db.execute(
+            sa.select(models.DatabaseUserTokenModel).where(
+                models.DatabaseUserTokenModel.user_id == user_id
+            )
+        )
+        .scalars()
+        .all()
+    )
+
+
+def delete_token(
+    db: orm.Session, existing_token: models.DatabaseUserTokenModel
+) -> models.DatabaseUserTokenModel:
+    db.delete(existing_token)
+    db.commit()
+    return existing_token

backend/capellacollab/users/tokens/models.py

@@ -0,0 +1,21 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+
+import sqlalchemy as sa
+from sqlalchemy import orm
+
+from capellacollab.core import database
+
+
+class DatabaseUserTokenModel(database.Base):
+    __tablename__ = "basic_auth_token"
+
+    id: orm.Mapped[int] = orm.mapped_column(
+        primary_key=True, index=True, autoincrement=True
+    )
+    user_id: orm.Mapped[int] = orm.mapped_column(sa.ForeignKey("users.id"))
+    hash: orm.Mapped[str]
+    expiration_date: orm.Mapped[datetime.date]
+    description: orm.Mapped[str]

backend/capellacollab/users/tokens/routes.py

@@ -0,0 +1,58 @@
+# SPDX-FileCopyrightText: Copyright DB Netz AG and the capella-collab-manager contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import fastapi
+from capellacollab.core import database
+from capellacollab.core.authentication import injectables as auth_injectables
+from capellacollab.users import injectables as user_injectables
+from capellacollab.users import models as users_models
+from sqlalchemy import orm
+
+from . import crud
+
+router = fastapi.APIRouter(
+    dependencies=[
+        fastapi.Depends(
+            auth_injectables.RoleVerification(
+                required_role=users_models.Role.USER
+            )
+        )
+    ]
+)
+
+
+@router.post("/current/token")
+def create_token_for_user(
+    user: users_models.DatabaseUser = fastapi.Depends(
+        user_injectables.get_own_user
+    ),
+    db: orm.Session = fastapi.Depends(database.get_db),
+    description: str = fastapi.Body(),
+):
+    _, password = crud.create_token(db, user.id, description)
+    return password
+
+
+@router.get("/current/tokens")
+def get_all_token_of_user(
+    user: users_models.DatabaseUser = fastapi.Depends(
+        user_injectables.get_own_user
+    ),
+    db: orm.Session = fastapi.Depends(database.get_db),
+):
+    return crud.get_token_by_user(db, user.id)
+
+
+@router.delete("/current/token/{id}")
+def delete_token_for_user(
+    id: int,
+    user: users_models.DatabaseUser = fastapi.Depends(
+        user_injectables.get_own_user
+    ),
+    db: orm.Session = fastapi.Depends(database.get_db),
+):
+    token_list = crud.get_token_by_user(db, user.id)
+    if token_list:
+        token = [token for token in token_list if token.id == id][0]
+        return crud.delete_token(db, token)
+    return None

backend/pyproject.toml

@@ -44,6 +44,7 @@ dependencies = [
   "starlette-prometheus",
   "fastapi-pagination>=0.12.5",
   "aiohttp",
+  "argon2-cffi",
 ]
 
 [project.urls]