Add basic auth

src/routers/member.py

@@ -1,34 +1,44 @@
 """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 ..models.member import Member
+from ..security import authorize
 from ..services.member import MemberService
 
 router = APIRouter(tags=["Member"], prefix="/api/v1/member")
 service: MemberService = MemberService()
 
 
 @router.post("", response_model=Member)
-def create_member_folder(member: Member):
+def create_member_folder(
+    member: Member,
+    authorized: Annotated[None, Depends(authorize)],  # pylint: disable=unused-argument
+):
     """
     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)],  # pylint: disable=unused-argument
+):
     """
     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 +48,19 @@ 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)],  # pylint: disable=unused-argument
+):
     """
     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,34 +1,44 @@
 """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 ..models.video import Video
+from ..security import authorize
 from ..services.video import VideoService
 
 router = APIRouter(tags=["Video"], prefix="/api/v1/video")
 service: VideoService = VideoService()
 
 
 @router.post("", response_model=Video)
-def create_video_folder(video: Video):
+def create_video_folder(
+    video: Video,
+    authorized: Annotated[None, Depends(authorize)],  # pylint: disable=unused-argument
+):
     """
     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)],  # pylint: disable=unused-argument
+):
     """
     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 +48,19 @@ 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)],  # pylint: disable=unused-argument
+):
     """
     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,36 @@
+"""Security module for the FastAPI application."""
+
+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)]):
+    """
+    Authorize the request with the correct username and password.
+    The correct username and password are stored in the settings.
+    :param credentials: the credentials from the request
+    :return:
+    """
+    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/routers/test_member.py

@@ -22,12 +22,14 @@ def test_create_member_folder(client, mocker):
 
     member_data = {"id": str(member_object.id), "url": member_object.url}
 
-    response = client.post("/api/v1/member", json=member_data)
+    response = client.post(
+        "/api/v1/member", json=member_data, auth=("admin", "password")
+    )
 
-    assert service_mock.create_folder_structure.call_count == 1
-    assert service_mock.create_folder_structure.call_args[0][0] == member_object
     assert response.status_code == 200
     assert response.json() == member_data
+    assert service_mock.create_folder_structure.call_count == 1
+    assert service_mock.create_folder_structure.call_args[0][0] == member_object
 
 
 def test_update_member_folder_no_id(client, mocker):
@@ -36,7 +38,9 @@ def test_update_member_folder_no_id(client, mocker):
 
     member_data = {"id": str(member_object.id), "url": member_object.url}
 
-    response = client.put("/api/v1/member", json=member_data)
+    response = client.put(
+        "/api/v1/member", json=member_data, auth=("admin", "password")
+    )
 
     assert service_mock.update_symlink.call_count == 0
     assert response.status_code == 404
@@ -49,7 +53,9 @@ def test_update_member_folder(client, mocker):
 
     member_data = {"id": str(member_object.id), "url": member_object.url}
 
-    response = client.put("/api/v1/member", json=member_data)
+    response = client.put(
+        "/api/v1/member", json=member_data, auth=("admin", "password")
+    )
 
     assert service_mock.update_symlink.call_count == 1
     assert service_mock.update_symlink.call_args[0][0] == member_object
@@ -64,6 +70,7 @@ def test_upload_member_picture_no_id(client, mocker):
     response = client.post(
         f"/api/v1/member/{member_object.id}/profilePicture",
         files={"file": ("file.jpg", "file_content", "image/jpeg")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_profile_picture.call_count == 0
@@ -77,6 +84,7 @@ def test_upload_member_picture_not_image(client, mocker):
     response = client.post(
         f"/api/v1/member/{member_object.id}/profilePicture",
         files={"file": ("file.jpg", "file_content", "text/plain")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_profile_picture.call_count == 0
@@ -92,6 +100,7 @@ def test_upload_member_picture(client, mocker):
     response = client.post(
         f"/api/v1/member/{member_object.id}/profilePicture",
         files={"file": ("file.jpg", "file_content", "image/jpeg")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_profile_picture.call_count == 1

tests/routers/test_video.py

@@ -24,7 +24,7 @@ def test_create_video_folder(client, mocker):
 
     video_data = {"id": str(video_object.id), "urls": video_object.urls}
 
-    response = client.post("/api/v1/video", json=video_data)
+    response = client.post("/api/v1/video", json=video_data, auth=("admin", "password"))
 
     assert service_mock.create_folder_structure.call_count == 1
     assert service_mock.create_folder_structure.call_args[0][0] == video_object
@@ -38,7 +38,7 @@ def test_update_video_folder_no_id(client, mocker):
 
     video_data = {"id": str(video_object.id), "urls": video_object.urls}
 
-    response = client.put("/api/v1/video", json=video_data)
+    response = client.put("/api/v1/video", json=video_data, auth=("admin", "password"))
 
     assert service_mock.update_symlinks.call_count == 0
     assert response.status_code == 404
@@ -51,7 +51,7 @@ def test_update_video_folder(client, mocker):
 
     video_data = {"id": str(video_object.id), "urls": video_object.urls}
 
-    response = client.put("/api/v1/video", json=video_data)
+    response = client.put("/api/v1/video", json=video_data, auth=("admin", "password"))
 
     assert service_mock.update_symlinks.call_count == 1
     assert service_mock.update_symlinks.call_args[0][0] == video_object
@@ -66,6 +66,7 @@ def test_upload_video_poster_no_id(client, mocker):
     response = client.post(
         f"/api/v1/video/{video_object.id}/thumbnail",
         files={"file": ("file.jpg", "file_content", "text/plain")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_thumbnail.call_count == 0
@@ -79,6 +80,7 @@ def test_upload_video_poster_not_image(client, mocker):
     response = client.post(
         f"/api/v1/video/{video_object.id}/thumbnail",
         files={"file": ("file.jpg", b"file_content", "text/plain")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_thumbnail.call_count == 0
@@ -94,6 +96,7 @@ def test_upload_video_poster(client, mocker):
     response = client.post(
         f"/api/v1/video/{video_object.id}/thumbnail",
         files={"file": ("file.jpg", b"file_content", "image/jpeg")},
+        auth=("admin", "password"),
     )
 
     assert service_mock.create_thumbnails.call_count == 1

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"