add basic auth

Author: csikb

src/routers/member.py

@@ -1,9 +1,12 @@
 """Member endpoints."""
 
 import re
+from typing import Annotated
 from uuid import UUID
 
-from fastapi import APIRouter, Response, UploadFile, status
+from fastapi import APIRouter, Depends, Response, UploadFile, status
+
+from security import authorize
 
 from ..models.member import Member
 from ..services.member import MemberService
@@ -13,22 +16,28 @@
 
 
 @router.post("", response_model=Member)
-def create_member_folder(member: Member):
+def create_member_folder(
+    member: Member, authorized: Annotated[None, Depends(authorize)]
+):
     """
     Create a folder structure for a member and return the member object.
     :param member: Member object
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original member object
     """
     service.create_folder_structure(member)
     return member
 
 
 @router.put("", response_model=Member)
-def update_member_folder(member: Member):
+def update_member_folder(
+    member: Member, authorized: Annotated[None, Depends(authorize)]
+):
     """
     Update the folder structure for a member and return the member object.
     If the member does not exist, return a 404.
     :param member: Member object
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original member object
     """
     if not service.to_id_path(member.id).exists():
@@ -38,14 +47,17 @@ def update_member_folder(member: Member):
 
 
 @router.post("/{member_id}/profilePicture", response_model=UUID)
-async def upload_member_picture(member_id: UUID, file: UploadFile):
+async def upload_member_picture(
+    member_id: UUID, file: UploadFile, authorized: Annotated[None, Depends(authorize)]
+):
     """
     Upload a picture for a member to convert
     and store the profile picture in different formats
     If the member does not exist, return a 404.
     If the file is not an image, return a 500.
     :param member_id: the id of the member
     :param file: the image file
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original member_id
     """
     # pylint: disable=duplicate-code

src/routers/video.py

@@ -1,9 +1,12 @@
 """Video endpoints"""
 
 import re
+from typing import Annotated
 from uuid import UUID
 
-from fastapi import APIRouter, Response, UploadFile, status
+from fastapi import APIRouter, Depends, Response, UploadFile, status
+
+from security import authorize
 
 from ..models.video import Video
 from ..services.video import VideoService
@@ -13,22 +16,24 @@
 
 
 @router.post("", response_model=Video)
-def create_video_folder(video: Video):
+def create_video_folder(video: Video, authorized: Annotated[None, Depends(authorize)]):
     """
     Create a folder structure for a video and return the video object.
     :param video: Video object
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original video object
     """
     service.create_folder_structure(video)
     return video
 
 
 @router.put("", response_model=Video)
-def update_video_folder(video: Video):
+def update_video_folder(video: Video, authorized: Annotated[None, Depends(authorize)]):
     """
     Update the folder structure for a video and return the video object.
     If the video does not exist, return a 404.
     :param video: Video object
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original video object
     """
     if not service.to_id_path(video.id).exists():
@@ -38,14 +43,17 @@ def update_video_folder(video: Video):
 
 
 @router.post("/{video_id}/thumbnail", response_model=UUID)
-async def upload_video_poster(video_id: UUID, file: UploadFile):
+async def upload_video_poster(
+    video_id: UUID, file: UploadFile, authorized: Annotated[None, Depends(authorize)]
+):
     """
     Upload a picture for a video thumbnail to convert
     and store the thumbnail in different formats
     If the video does not exist, return a 404.
     If the file is not an image, return a 500.
     :param video_id: the id of the video
     :param file: the image file
+    :param authorized: fastapi dependency to authorize the request
     :return: 200 and the original video_id
     """
     # pylint: disable=duplicate-code

src/security.py

@@ -0,0 +1,28 @@
+import secrets
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+
+from settings import settings
+
+security = HTTPBasic()
+
+
+def authorize(credentials: Annotated[HTTPBasicCredentials, Depends(security)]):
+    current_username_bytes = credentials.username.encode("utf8")
+    correct_username_bytes = settings.username.encode("utf8")
+    is_correct_username = secrets.compare_digest(
+        current_username_bytes, correct_username_bytes
+    )
+    current_password_bytes = credentials.password.encode("utf8")
+    correct_password_bytes = settings.password.encode("utf8")
+    is_correct_password = secrets.compare_digest(
+        current_password_bytes, correct_password_bytes
+    )
+    if not (is_correct_username and is_correct_password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Basic"},
+        )

src/settings.py

@@ -7,6 +7,8 @@ class Settings(BaseSettings):
     """Class for reading settings from environment variables."""
 
     server_base_path: str = "./assets/"
+    username: str = "admin"
+    password: str = "password"
 
 
 settings = Settings()

tests/test_security.py

@@ -0,0 +1,37 @@
+import pytest
+from fastapi import Depends, FastAPI
+from fastapi.security import HTTPBasicCredentials
+from fastapi.testclient import TestClient
+
+from src.security import authorize
+
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+
+    @app.get("/secure-endpoint")
+    def secure_endpoint(credentials: HTTPBasicCredentials = Depends(authorize)):
+        return {"message": "You are authorized"}
+
+    return TestClient(app)
+
+
+def test_authorize_correct_credentials(mocker, client):
+    credentials = HTTPBasicCredentials(username="admin", password="password")
+    mocker.patch("src.security.security", return_value=credentials)
+
+    response = client.get("/secure-endpoint", auth=("admin", "password"))
+
+    assert response.status_code == 200
+    assert response.json() == {"message": "You are authorized"}
+
+
+def test_authorize_incorrect_credentials(mocker, client):
+    credentials = HTTPBasicCredentials(username="admin", password="password")
+    mocker.patch("src.security.security", return_value=credentials)
+
+    response = client.get("/secure-endpoint", auth=("wrong", "wrong"))
+
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Incorrect username or password"}

tests/test_settings.py

@@ -0,0 +1,8 @@
+from src.settings import Settings
+
+
+def test_settings_default_values():
+    settings = Settings()
+    assert settings.server_base_path == "./assets/"
+    assert settings.username == "admin"
+    assert settings.password == "password"