Backend authentication [gh-6] (#16)

* feat: user creation, use username as identifier, admin account creation, configure auth in env

* chore: linter

* chore: linter

* chore: .envexample update

* chore: update poetry lock

Author: mateuszjasinski

.envexample

@@ -4,8 +4,12 @@ POSTGRES_USER: dev
 POSTGRES_PASSWORD: local-dev
 POSTGRES_DB: cpf
 
+BASE_URL: /cpf/api
+USE_MOCK_USER: False
 
 FUSIONAUTH_POSTGRES_DB: fusionauth
 FUSIONAUTH_POSTGRES_USER: fusionauth
 FUSIONAUTH_POSTGRES_PASSWORD: password
-
+FUSIONAUTH_API_KEY: 00c24029-e66a-43f2-9000-544e90e8d46e
+FUSIONAUTH_URL: http://fusionauth:9011
+FUSIONAUTH_APPLICATION_ID: 23e4b229-1219-42e5-aed6-f9b6f1eedef8

backend/src/cpf/adapters/inbound/data_loader/loader.py

@@ -1,12 +1,13 @@
 import json
 import os
 
-from cpf.core.ports.provided.services import ManageService
+from cpf.core.ports.provided.services import ManageService, UserManagementService
 
 manage_service: ManageService | None = None
+user_management_service: UserManagementService | None = None
 
 
-def set_manage_service(service: ManageService):
+def set_manage_service(service: ManageService) -> None:
     global manage_service
     manage_service = service
 
@@ -17,6 +18,17 @@ def get_manage_service() -> ManageService:
     return manage_service
 
 
+def set_user_management_service(service: UserManagementService) -> None:
+    global user_management_service
+    user_management_service = service
+
+
+def get_user_management_service() -> UserManagementService:
+    if not user_management_service:
+        raise RuntimeError("User management service not set")
+    return user_management_service
+
+
 def start_data_upload() -> None:
     service = get_manage_service()
     if service.check_if_data_is_exists():
@@ -38,3 +50,7 @@ def start_data_upload() -> None:
             ladder_data = json.loads(file.read())
             service.create_ladder(ladder_data=ladder_data)
             print(f"Ladder created from file {ladder_data_path}")
+    # Create admin account
+    user_service = get_user_management_service()
+    # TODO Create new service method for admin account creation when role management system will be ready
+    user_service.create_new_user(email="[email protected]", first_name="Cpf", last_name="Admin")

backend/src/cpf/adapters/inbound/rest_api/library/api.py

@@ -1,4 +1,5 @@
 import os
+from typing import Annotated
 
 from fastapi import APIRouter, Depends
 
@@ -13,16 +14,19 @@
     LadderDetailResponse,
     LadderResponse,
 )
-from cpf.adapters.inbound.rest_api.rest_api import get_library_query_service
+from cpf.adapters.inbound.rest_api.permissions import check_permissions
+from cpf.adapters.inbound.rest_api.rest_api import auth, get_library_query_service
 from cpf.core.ports.provided.services import QueryService
-from cpf.core.ports.required.dtos import LadderDetailDTO
+from cpf.core.ports.required.dtos import LadderDetailDTO, UserDTO
 from cpf.core.ports.required.readmodels import BucketReadModel, LadderReadModel
 
 router = APIRouter(prefix=f"{os.getenv('BASE_URL')}/library")
 
 
 @router.get(path="/ladders", response_model_exclude_none=True)
+@check_permissions(permission_classes=[])
 def get_ladders(
+    user: Annotated[UserDTO | None, Depends(auth)],
     service: QueryService = Depends(get_library_query_service),
 ) -> list[LadderResponse]:
     ladder_read_models: list[LadderReadModel] = service.get_all_ladders()
@@ -37,8 +41,11 @@ def get_ladders(
 
 
 @router.get(path="/ladders/{ladder_slug}", response_model_exclude_none=True)
+@check_permissions(permission_classes=[])
 def get_ladder_details(
-    ladder_slug: str, service: QueryService = Depends(get_library_query_service)
+    ladder_slug: str,
+    user: Annotated[UserDTO | None, Depends(auth)],
+    service: QueryService = Depends(get_library_query_service),
 ) -> LadderDetailResponse:
     ladder_detail: LadderDetailDTO = service.get_ladder(ladder_slug=ladder_slug)
 
@@ -64,8 +71,11 @@ def get_ladder_details(
 
 
 @router.get(path="/buckets/{bucket_slug}", response_model_exclude_none=True)
+@check_permissions(permission_classes=[])
 def get_bucket_details(
-    bucket_slug: str, service: QueryService = Depends(get_library_query_service)
+    bucket_slug: str,
+    user: Annotated[UserDTO | None, Depends(auth)],
+    service: QueryService = Depends(get_library_query_service),
 ) -> BucketDetailResponse:
     bucket_details: BucketReadModel = service.get_bucket(bucket_slug=bucket_slug)
     advancement_levels: list[AdvancementLevelResponse] = []

backend/src/cpf/adapters/inbound/rest_api/models/responses/core.py

@@ -1,5 +1,20 @@
 from cpf.adapters.inbound.rest_api.ion import IonBaseModel, IonLink
 
 
-class RootResponse(IonBaseModel):
+class UserResponse(IonBaseModel):
+    first_name: str
+    last_name: str
+
+
+class UnauthenticatedRootResponse(IonBaseModel):
+    login: IonLink
+
+
+class AuthenticatedRootResponse(IonBaseModel):
+    user: UserResponse
     get_ladders: IonLink
+
+
+class LadderResponse(IonBaseModel):
+    ladder_name: str
+    ladder_slug: str

backend/src/cpf/adapters/inbound/rest_api/permissions.py

@@ -0,0 +1,31 @@
+from abc import ABC, abstractmethod
+from functools import wraps
+
+from fastapi import HTTPException
+from starlette import status
+
+from cpf.core.ports.required.dtos import UserDTO
+
+
+class Permission(ABC):
+
+    @classmethod
+    @abstractmethod
+    def validate_permission(cls, user: UserDTO) -> bool:
+        pass
+
+
+def check_permissions(permission_classes: list[type[Permission]]):
+    def inner(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            user: UserDTO = kwargs.get("user")
+            if not user:
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User not authenticated")
+            if not all(permission.validate_permission(user=user) for permission in permission_classes):
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User has no permission")
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return inner

backend/src/cpf/adapters/inbound/rest_api/rest_api.py

@@ -1,14 +1,29 @@
-from fastapi import APIRouter, FastAPI
+import os
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, FastAPI
+from starlette.requests import Request
 
 from cpf.adapters.inbound.rest_api.ion import IonLink
-from cpf.adapters.inbound.rest_api.models.responses.core import RootResponse
-from cpf.core.ports.provided.services import ManageService, QueryService
+from cpf.adapters.inbound.rest_api.models.responses.core import (
+    AuthenticatedRootResponse,
+    UnauthenticatedRootResponse,
+    UserResponse,
+)
+from cpf.adapters.inbound.rest_api.utils import env_to_bool, fake_user_factory
+from cpf.core.ports.provided.services import (
+    ManageService,
+    QueryService,
+    UserManagementService,
+)
+from cpf.core.ports.required.dtos import UserDTO
 
 router = APIRouter(prefix="/cpf/api")
 app = FastAPI()
 
 library_manage_service: ManageService | None = None
 library_query_service: QueryService | None = None
+user_management_service: UserManagementService | None = None
 
 
 def set_library_manage_service(service: ManageService):
@@ -33,9 +48,47 @@ def get_library_query_service() -> QueryService:
     return library_query_service
 
 
-@router.get(path="")
-def get_api_root() -> RootResponse:
-    return RootResponse(get_ladders=IonLink(href="/cpf/api/library/ladders/"))
+def set_user_management_service(service: UserManagementService) -> None:
+    global user_management_service
+    user_management_service = service
+
+
+def get_user_management_service() -> UserManagementService:
+    if not user_management_service:
+        raise RuntimeError("User management service not set")
+    return user_management_service
+
+
+class FastAPIAuth:
+
+    def __call__(self, request: Request) -> UserDTO:
+        # TODO Remove after auth will be implemented on frontend
+        if env_to_bool(os.getenv("USE_MOCK_USER")):
+            return fake_user_factory()
+        service_instance: UserManagementService = get_user_management_service()
+        return service_instance.get_user(access_token=request.cookies.get("access_token"))
+
+
+auth = FastAPIAuth()
+
+
+@router.get(path="", response_model_exclude_none=True)
+def get_api_root(
+    user: Annotated[UserDTO | None, Depends(auth)]
+) -> AuthenticatedRootResponse | UnauthenticatedRootResponse:
+    if not user:
+        return UnauthenticatedRootResponse(
+            # TODO Create social auth login redirect
+            login=IonLink(href="/api/login")
+        )
+
+    return AuthenticatedRootResponse(
+        user=UserResponse(
+            first_name=user.first_name,
+            last_name=user.last_name,
+        ),
+        get_ladders=IonLink(href="/cpf/api/library/ladders"),
+    )
 
 
 @router.get(path="/health")
@@ -44,6 +97,8 @@ def health_check():
 
 
 from cpf.adapters.inbound.rest_api.library.api import router as library_router  # noqa
+from cpf.adapters.inbound.rest_api.users.api import router as users_router  # noqa
 
 app.include_router(router=router)
 app.include_router(router=library_router)
+app.include_router(router=users_router)

backend/src/cpf/adapters/inbound/rest_api/users/__init__.py

@@ -0,0 +1,32 @@
+import os
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+
+from cpf.adapters.inbound.rest_api.permissions import check_permissions
+from cpf.core.ports.provided.services import UserManagementService
+from cpf.core.ports.required.dtos import UserDTO
+
+from ..rest_api import auth, get_user_management_service
+from .models.requests import PutUser
+from .models.responses import UserResponse
+
+router = APIRouter(prefix=f"{os.getenv('BASE_URL')}/users")
+
+
+@router.post(path="", response_model_exclude_none=True)
+@check_permissions(permission_classes=[])
+def create_new_user(
+    request: PutUser,
+    user: Annotated[UserDTO, Depends(auth)],
+    service: UserManagementService = Depends(get_user_management_service),
+) -> UserResponse:
+    new_user: UserDTO = service.create_new_user(
+        first_name=request.first_name, last_name=request.last_name, email=request.email
+    )
+    return UserResponse(
+        username=new_user.username,
+        email=new_user.email,
+        first_name=new_user.first_name,
+        last_name=new_user.last_name,
+    )

backend/src/cpf/adapters/inbound/rest_api/users/api.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel, EmailStr, Field
+
+
+class PutUser(BaseModel):
+    first_name: str = Field(..., description="New user first name")
+    last_name: str = Field(..., description="New user last name")
+    email: EmailStr = Field(..., description="New user email")

backend/src/cpf/adapters/inbound/rest_api/users/models/__init__.py

@@ -0,0 +1,8 @@
+from cpf.adapters.inbound.rest_api.ion import IonBaseModel
+
+
+class UserResponse(IonBaseModel):
+    username: str
+    email: str
+    first_name: str
+    last_name: str

backend/src/cpf/adapters/inbound/rest_api/users/models/requests.py

@@ -3,6 +3,16 @@
 from fastapi import FastAPI
 from starlette.routing import Route
 
+from cpf.core.ports.required.dtos import UserDTO
+
+
+def env_to_bool(value: str) -> bool:
+    return value.lower() in ("true", "1")
+
+
+def fake_user_factory() -> UserDTO:
+    return UserDTO(identifier="[email protected]", first_name="Mock", last_name="John")
+
 
 class EndpointInfo(TypedDict):
     method: str | None

backend/src/cpf/adapters/inbound/rest_api/users/models/responses.py

@@ -0,0 +1,56 @@
+import os
+
+from fastapi import HTTPException
+from fusionauth.fusionauth_client import FusionAuthClient
+from starlette import status
+
+from cpf.core.ports.required.clients import AuthenticationClient
+from cpf.core.ports.required.readmodels import UserReadModel
+
+
+class FusionAuthAuthenticationClient(AuthenticationClient):
+
+    def __init__(self, client: FusionAuthClient):
+        self._client = client
+
+    @staticmethod
+    def _get_user_read_model_from_response(response_data: dict) -> UserReadModel:
+        user_data = response_data.get("user")
+        return UserReadModel(
+            username=user_data.get("username"),
+            email=user_data.get("email"),
+            first_name=user_data.get("data").get("first_name"),
+            last_name=user_data.get("data").get("last_name"),
+        )
+
+    def get_user_data(self, access_token: str) -> UserReadModel | None:
+        fusion_auth_response = self._client.retrieve_user_using_jwt(encoded_jwt=access_token)
+        if fusion_auth_response.status != status.HTTP_200_OK:
+            return None
+
+        return self._get_user_read_model_from_response(fusion_auth_response.success_response)
+
+    def create_user(self, email: str, username: str, first_name: str, last_name: str) -> UserReadModel:
+        user_request = {
+            "user": {
+                "email": email,
+                "username": username,
+                "password": "dev-use-only",
+                "data": {
+                    "first_name": first_name,
+                    "last_name": last_name,
+                },
+            }
+        }
+        client_response = self._client.create_user(request=user_request, user_id=None)
+        if not client_response.was_successful():
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
+
+        return self._get_user_read_model_from_response(client_response.success_response)
+
+
+fusion_auth_client = FusionAuthClient(api_key=os.getenv("FUSIONAUTH_API_KEY"), base_url=os.getenv("FUSIONAUTH_URL"))
+
+
+def authentication_client_factory() -> AuthenticationClient:
+    return FusionAuthAuthenticationClient(client=fusion_auth_client)

backend/src/cpf/adapters/inbound/rest_api/utils.py

@@ -50,7 +50,7 @@ def save(self, uuid: str, serialize_data: dict):
                         serialize_data=serialize_data,
                     )
                 else:
-                    cursor = self._execute_update(cursor=cursor, ladder_slug=uuid, serialize_data=serialize_data)
+                    cursor = self._execute_update(cursor=cursor, bucket_slug=uuid, serialize_data=serialize_data)
                 conn.commit()
 
     def check_if_bucket_exists(self, bucket_slug: str) -> bool: