From d4b8a95a93cdb15374f312693cce500cdb3f1e60 Mon Sep 17 00:00:00 2001 From: AbiramiR-27 Date: Sat, 25 Jul 2026 08:49:58 +0530 Subject: [PATCH 1/5] fix(backend): prevent path traversal in face search endpoint (fixes #1322) --- backend/app/routes/face_clusters.py | 32 +++++++++++++++++++++++++++++ backend/tests/test_face_clusters.py | 31 ++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py index 4951ac960..b340f306b 100644 --- a/backend/app/routes/face_clusters.py +++ b/backend/app/routes/face_clusters.py @@ -33,11 +33,34 @@ ) from app.schemas.images import FaceSearchRequest, InputType from app.utils.faceSearch import perform_face_search +from app.database.folders import db_get_all_folders logger = logging.getLogger(__name__) router = APIRouter() +def is_safe_path(target_path: str) -> bool: + """Validate that target_path is within one of the registered folders or temp_uploads.""" + + abs_target = os.path.abspath(target_path) + try: + allowed_folders = db_get_all_folders() + except Exception: + allowed_folders = [] + + temp_dir = os.path.abspath("temp_uploads") + allowed_folders.append(temp_dir) + + for folder in allowed_folders: + abs_folder = os.path.abspath(folder) + try: + if os.path.commonpath([abs_folder, abs_target]) == abs_folder: + return True + except ValueError: + continue + return False + + @router.put( "/{cluster_id}", response_model=RenameClusterResponse, @@ -245,6 +268,15 @@ def face_tagging( message="The provided path is not a valid file", ).model_dump(), ) + if not is_safe_path(local_file_path): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ErrorResponse( + success=False, + error="Access Denied", + message="Access to the specified file path is restricted.", + ).model_dump(), + ) image_path = payload.path elif input_type == InputType.base64: diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index 4f5d7d703..68d71aadd 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -417,6 +417,37 @@ def test_unsupported_http_methods(self, method, endpoint): response = client.request(method, endpoint) assert response.status_code == 405 + @patch("app.routes.face_clusters.db_get_all_folders") + @patch("app.routes.face_clusters.perform_face_search") + def test_face_search_path_traversal_blocked(self, mock_perform, mock_folders): + """Test that paths outside allowed folders are blocked with 403.""" + mock_folders.return_value = ["/allowed/folder"] + mock_perform.return_value = {"success": True, "data": []} + + # Mock os.path.isfile to simulate a valid file target + with patch("os.path.isfile", return_value=True): + response = client.post( + "/face_clusters/face-search?input_type=path", + json={"path": "/restricted/file.jpg", "base64_data": ""}, + ) + assert response.status_code == 403 + assert response.json()["detail"]["error"] == "Access Denied" + + @patch("app.routes.face_clusters.db_get_all_folders") + @patch("app.routes.face_clusters.perform_face_search") + def test_face_search_safe_path_allowed(self, mock_perform, mock_folders): + """Test that paths inside allowed folders are allowed.""" + mock_folders.return_value = ["/allowed/folder"] + mock_perform.return_value = {"success": True, "data": []} + + with patch("os.path.isfile", return_value=True): + response = client.post( + "/face_clusters/face-search?input_type=path", + json={"path": "/allowed/folder/family.jpg", "base64_data": ""}, + ) + assert response.status_code == 200 + mock_perform.assert_called_once_with("/allowed/folder/family.jpg") + # ============================================================================ # Algorithmic Logic Tests From 722a61bd4b7e72bfa591144917bcd8585ce0453e Mon Sep 17 00:00:00 2001 From: AbiramiR-27 Date: Sat, 25 Jul 2026 09:12:36 +0530 Subject: [PATCH 2/5] fix(backend): resolve canonical paths and prevent file existence disclosures in face-search (fixes #1322) --- backend/app/routes/face_clusters.py | 24 ++++++++-------- backend/tests/test_face_clusters.py | 44 ++++++++++++++++++++++++++--- 2 files changed, 52 insertions(+), 16 deletions(-) diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py index b340f306b..7ff48ece6 100644 --- a/backend/app/routes/face_clusters.py +++ b/backend/app/routes/face_clusters.py @@ -42,19 +42,19 @@ def is_safe_path(target_path: str) -> bool: """Validate that target_path is within one of the registered folders or temp_uploads.""" - abs_target = os.path.abspath(target_path) + real_target = os.path.realpath(target_path) try: allowed_folders = db_get_all_folders() except Exception: allowed_folders = [] - temp_dir = os.path.abspath("temp_uploads") + temp_dir = os.path.realpath("temp_uploads") allowed_folders.append(temp_dir) for folder in allowed_folders: - abs_folder = os.path.abspath(folder) + real_folder = os.path.realpath(folder) try: - if os.path.commonpath([abs_folder, abs_target]) == abs_folder: + if os.path.commonpath([real_folder, real_target]) == real_folder: return True except ValueError: continue @@ -259,22 +259,22 @@ def face_tagging( message="image path is required.", ).model_dump(), ) - if not os.path.isfile(local_file_path): + if not is_safe_path(local_file_path): raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, + status_code=status.HTTP_403_FORBIDDEN, detail=ErrorResponse( success=False, - error="Invalid file path", - message="The provided path is not a valid file", + error="Access Denied", + message="Access to the specified file path is restricted.", ).model_dump(), ) - if not is_safe_path(local_file_path): + if not os.path.isfile(local_file_path): raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, + status_code=status.HTTP_400_BAD_REQUEST, detail=ErrorResponse( success=False, - error="Access Denied", - message="Access to the specified file path is restricted.", + error="Invalid file path", + message="The provided path is not a valid file", ).model_dump(), ) image_path = payload.path diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index 68d71aadd..47885f7e7 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -424,11 +424,47 @@ def test_face_search_path_traversal_blocked(self, mock_perform, mock_folders): mock_folders.return_value = ["/allowed/folder"] mock_perform.return_value = {"success": True, "data": []} - # Mock os.path.isfile to simulate a valid file target - with patch("os.path.isfile", return_value=True): + # Mock os.path.realpath to resolve the traversal path + def realpath_mock(path): + # Normalizing Windows drive prefixes or slashes for tests + cleaned = path.replace("\\", "/").lower() + if "../restricted" in cleaned or "/restricted" in cleaned: + return "/restricted/file.jpg" + if "/allowed/folder" in cleaned: + return "/allowed/folder" + return path + + with patch("os.path.realpath", side_effect=realpath_mock): + response = client.post( + "/face_clusters/face-search?input_type=path", + json={ + "path": "/allowed/folder/../restricted/file.jpg", + "base64_data": "", + }, + ) + assert response.status_code == 403 + assert response.json()["detail"]["error"] == "Access Denied" + + @patch("app.routes.face_clusters.db_get_all_folders") + @patch("app.routes.face_clusters.perform_face_search") + def test_face_search_symlink_escape_blocked(self, mock_perform, mock_folders): + """Test that symlinks inside allowed folders pointing outside are rejected.""" + mock_folders.return_value = ["/allowed/folder"] + mock_perform.return_value = {"success": True, "data": []} + + # Simulating a symlink at "/allowed/folder/link.jpg" pointing to "/restricted/secret.jpg" + def realpath_mock(path): + cleaned = path.replace("\\", "/").lower() + if cleaned == "/allowed/folder/link.jpg": + return "/restricted/secret.jpg" + if cleaned == "/allowed/folder": + return "/allowed/folder" + return path + + with patch("os.path.realpath", side_effect=realpath_mock): response = client.post( "/face_clusters/face-search?input_type=path", - json={"path": "/restricted/file.jpg", "base64_data": ""}, + json={"path": "/allowed/folder/link.jpg", "base64_data": ""}, ) assert response.status_code == 403 assert response.json()["detail"]["error"] == "Access Denied" @@ -446,7 +482,7 @@ def test_face_search_safe_path_allowed(self, mock_perform, mock_folders): json={"path": "/allowed/folder/family.jpg", "base64_data": ""}, ) assert response.status_code == 200 - mock_perform.assert_called_once_with("/allowed/folder/family.jpg") + mock_perform.assert_called_once() # ============================================================================ From c7c73d1f2567a8e3a7f34a6dc84bb0aa75e7bb4d Mon Sep 17 00:00:00 2001 From: AbiramiR-27 Date: Sat, 25 Jul 2026 09:20:44 +0530 Subject: [PATCH 3/5] test(backend): harden blocked-path tests in test_face_clusters.py --- backend/tests/test_face_clusters.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index 47885f7e7..be248870b 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -434,7 +434,9 @@ def realpath_mock(path): return "/allowed/folder" return path - with patch("os.path.realpath", side_effect=realpath_mock): + with patch("os.path.realpath", side_effect=realpath_mock), patch( + "os.path.isfile", return_value=True + ): response = client.post( "/face_clusters/face-search?input_type=path", json={ @@ -444,6 +446,7 @@ def realpath_mock(path): ) assert response.status_code == 403 assert response.json()["detail"]["error"] == "Access Denied" + mock_perform.assert_not_called() @patch("app.routes.face_clusters.db_get_all_folders") @patch("app.routes.face_clusters.perform_face_search") @@ -461,13 +464,16 @@ def realpath_mock(path): return "/allowed/folder" return path - with patch("os.path.realpath", side_effect=realpath_mock): + with patch("os.path.realpath", side_effect=realpath_mock), patch( + "os.path.isfile", return_value=True + ): response = client.post( "/face_clusters/face-search?input_type=path", json={"path": "/allowed/folder/link.jpg", "base64_data": ""}, ) assert response.status_code == 403 assert response.json()["detail"]["error"] == "Access Denied" + mock_perform.assert_not_called() @patch("app.routes.face_clusters.db_get_all_folders") @patch("app.routes.face_clusters.perform_face_search") From b4bc922716d8e7b5fae59c91605b1cfced3fde1f Mon Sep 17 00:00:00 2001 From: AbiramiR-27 Date: Wed, 29 Jul 2026 18:37:57 +0530 Subject: [PATCH 4/5] fix(backend): eliminate TOCTOU race in face search by loading files securely with O_NOFOLLOW --- backend/app/models/FaceDetector.py | 21 +++++++++++-- backend/app/routes/face_clusters.py | 48 +++++++++++++---------------- backend/app/utils/faceSearch.py | 12 ++++++-- backend/tests/test_face_clusters.py | 25 ++++++++++++++- 4 files changed, 73 insertions(+), 33 deletions(-) diff --git a/backend/app/models/FaceDetector.py b/backend/app/models/FaceDetector.py index 3d4a9f385..007a1ad48 100644 --- a/backend/app/models/FaceDetector.py +++ b/backend/app/models/FaceDetector.py @@ -1,6 +1,7 @@ # app/detectors/FaceDetector.py import cv2 +import numpy as np from app.models.FaceNet import FaceNet from app.utils.FaceNet import FaceNet_util_preprocess_image, FaceNet_util_get_model_path from app.utils.YOLO import YOLO_util_get_model_path @@ -13,6 +14,7 @@ PICTO_CLUSTERING_MIN_FACE_SIZE, ) from app.utils.face_quality import face_passes_quality_gate +from typing import Optional # Initialize logger logger = get_logger(__name__) @@ -29,10 +31,23 @@ def __init__(self): self._initialized = True logger.info("FaceDetector initialized with YOLO and FaceNet models.") - def detect_faces(self, image_id: str, image_path: str, forSearch: bool = False): - img = cv2.imread(image_path) + def detect_faces( + self, + image_id: str, + image_path: Optional[str] = None, + forSearch: bool = False, + image_bytes: Optional[bytes] = None, + ): + if image_bytes is not None: + img = cv2.imdecode(np.frombuffer(image_bytes, np.uint8), cv2.IMREAD_COLOR) + elif image_path is not None: + img = cv2.imread(image_path) + else: + logger.error("Neither image_path nor image_bytes was provided.") + return None + if img is None: - logger.error(f"Failed to load image: {image_path}") + logger.error(f"Failed to load image: {image_path or ''}") return None boxes, scores, class_ids = self.yolo_detector(img) diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py index 01faba72f..13eb72ffe 100644 --- a/backend/app/routes/face_clusters.py +++ b/backend/app/routes/face_clusters.py @@ -3,7 +3,6 @@ import base64 from concurrent.futures import CancelledError, Future, ProcessPoolExecutor from typing import Annotated -import uuid import os from fastapi import APIRouter, Depends, HTTPException, Query, status from app.database.face_clusters import ( @@ -290,8 +289,6 @@ def face_tagging( InputType, Query(description="Choose input type: 'path' or 'base64'") ] = InputType.path, ): - image_path = None - if input_type == InputType.path: local_file_path = payload.path @@ -322,7 +319,26 @@ def face_tagging( message="The provided path is not a valid file", ).model_dump(), ) - image_path = payload.path + + try: + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd_handle = os.open(local_file_path, flags) + with open(fd_handle, "rb") as f: + image_bytes = f.read() + except Exception as e: + logger.error(f"Failed to securely open path {local_file_path}: {e}") + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ErrorResponse( + success=False, + error="Access Denied", + message="Cannot read the specified path.", + ).model_dump(), + ) + + return perform_face_search(image_bytes=image_bytes) elif input_type == InputType.base64: base64_data = payload.base64_data @@ -358,29 +374,7 @@ def face_tagging( ).model_dump(), ) - format_match = ( - base64_data.split(";")[0].split("/")[-1] if ";" in base64_data else "jpeg" - ) - extension = ( - format_match - if format_match in ["jpeg", "jpg", "png", "gif", "webp"] - else "jpeg" - ) - image_id = str(uuid.uuid4())[:8] - temp_dir = "temp_uploads" - os.makedirs(temp_dir, exist_ok=True) - local_image_path = os.path.join(temp_dir, f"{image_id}.{extension}") - - with open(local_image_path, "wb") as f: - f.write(image_bytes) - - image_path = local_image_path - - try: - return perform_face_search(image_path) - finally: - if input_type == InputType.base64 and image_path and os.path.exists(image_path): - os.remove(image_path) + return perform_face_search(image_bytes=image_bytes) @router.post( diff --git a/backend/app/utils/faceSearch.py b/backend/app/utils/faceSearch.py index 385cce908..57d88a4a9 100644 --- a/backend/app/utils/faceSearch.py +++ b/backend/app/utils/faceSearch.py @@ -32,12 +32,15 @@ class GetAllImagesResponse(BaseModel): data: List[ImageData] -def perform_face_search(image_path: str) -> GetAllImagesResponse: +def perform_face_search( + image_path: Optional[str] = None, image_bytes: Optional[bytes] = None +) -> GetAllImagesResponse: """ Performs face detection, embedding generation, and similarity search. Args: image_path (str): Path to the image file to process. + image_bytes (bytes): Optional in-memory raw image bytes. Returns: GetAllImagesResponse: Search result containing matched images. @@ -50,7 +53,12 @@ def perform_face_search(image_path: str) -> GetAllImagesResponse: image_id = str(uuid.uuid4()) try: - result = fd.detect_faces(image_id, image_path, forSearch=True) + result = fd.detect_faces( + image_id, + image_path=image_path, + forSearch=True, + image_bytes=image_bytes, + ) except Exception as e: return GetAllImagesResponse( success=False, diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index be248870b..4cd35ad5d 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -482,7 +482,11 @@ def test_face_search_safe_path_allowed(self, mock_perform, mock_folders): mock_folders.return_value = ["/allowed/folder"] mock_perform.return_value = {"success": True, "data": []} - with patch("os.path.isfile", return_value=True): + from unittest.mock import mock_open + + with patch("os.path.isfile", return_value=True), patch( + "os.open", return_value=123 + ), patch("builtins.open", mock_open(read_data=b"fakeimagebytes")): response = client.post( "/face_clusters/face-search?input_type=path", json={"path": "/allowed/folder/family.jpg", "base64_data": ""}, @@ -490,6 +494,25 @@ def test_face_search_safe_path_allowed(self, mock_perform, mock_folders): assert response.status_code == 200 mock_perform.assert_called_once() + @patch("app.routes.face_clusters.perform_face_search") + def test_face_search_base64_allowed(self, mock_perform): + """Test that base64 images are processed entirely in memory and allowed.""" + mock_perform.return_value = {"success": True, "data": []} + + # 1x1 transparent GIF base64 + gif_b64 = "data:image/gif;base64,R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7" + response = client.post( + "/face_clusters/face-search?input_type=base64", + json={"path": "", "base64_data": gif_b64}, + ) + assert response.status_code == 200 + mock_perform.assert_called_once() + # Verify that perform_face_search was called with image_bytes (not image_path) + called_kwargs = mock_perform.call_args.kwargs + assert "image_bytes" in called_kwargs + assert called_kwargs["image_bytes"] is not None + assert called_kwargs.get("image_path") is None + # ============================================================================ # Algorithmic Logic Tests From c5dd0db185a1ba659cdcc534832446d6f27a63fc Mon Sep 17 00:00:00 2001 From: AbiramiR-27 Date: Wed, 29 Jul 2026 19:00:51 +0530 Subject: [PATCH 5/5] fix(backend): eliminate authorization-to-open race condition in face search --- backend/app/models/FaceDetector.py | 2 +- backend/app/routes/face_clusters.py | 139 +++++++++++++++++++++++----- backend/tests/test_face_clusters.py | 31 ++++++- 3 files changed, 142 insertions(+), 30 deletions(-) diff --git a/backend/app/models/FaceDetector.py b/backend/app/models/FaceDetector.py index 007a1ad48..ee907537a 100644 --- a/backend/app/models/FaceDetector.py +++ b/backend/app/models/FaceDetector.py @@ -37,7 +37,7 @@ def detect_faces( image_path: Optional[str] = None, forSearch: bool = False, image_bytes: Optional[bytes] = None, - ): + ) -> Optional[dict]: if image_bytes is not None: img = cv2.imdecode(np.frombuffer(image_bytes, np.uint8), cv2.IMREAD_COLOR) elif image_path is not None: diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py index 13eb72ffe..b8c23a752 100644 --- a/backend/app/routes/face_clusters.py +++ b/backend/app/routes/face_clusters.py @@ -2,7 +2,7 @@ from binascii import Error as Base64Error import base64 from concurrent.futures import CancelledError, Future, ProcessPoolExecutor -from typing import Annotated +from typing import Annotated, Optional import os from fastapi import APIRouter, Depends, HTTPException, Query, status from app.database.face_clusters import ( @@ -42,9 +42,10 @@ router = APIRouter() -def is_safe_path(target_path: str) -> bool: - """Validate that target_path is within one of the registered folders or temp_uploads.""" - +def get_safe_root(target_path: str) -> Optional[str]: + """Validate that target_path is within one of the registered folders or temp_uploads + and return the matching resolved folder root. + """ real_target = os.path.realpath(target_path) try: allowed_folders = db_get_all_folders() @@ -58,10 +59,15 @@ def is_safe_path(target_path: str) -> bool: real_folder = os.path.realpath(folder) try: if os.path.commonpath([real_folder, real_target]) == real_folder: - return True + return real_folder except ValueError: continue - return False + return None + + +def is_safe_path(target_path: str) -> bool: + """Validate that target_path is within one of the registered folders or temp_uploads.""" + return get_safe_root(target_path) is not None def _log_rescore_outcome(cluster_id: str, done: "Future[int]") -> None: @@ -301,7 +307,8 @@ def face_tagging( message="image path is required.", ).model_dump(), ) - if not is_safe_path(local_file_path): + allowed_root = get_safe_root(local_file_path) + if not allowed_root: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ErrorResponse( @@ -320,23 +327,107 @@ def face_tagging( ).model_dump(), ) - try: - flags = os.O_RDONLY - if hasattr(os, "O_NOFOLLOW"): - flags |= os.O_NOFOLLOW - fd_handle = os.open(local_file_path, flags) - with open(fd_handle, "rb") as f: - image_bytes = f.read() - except Exception as e: - logger.error(f"Failed to securely open path {local_file_path}: {e}") - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail=ErrorResponse( - success=False, - error="Access Denied", - message="Cannot read the specified path.", - ).model_dump(), - ) + from pathlib import Path + + canonical_root = os.path.realpath(allowed_root) + canonical_target = os.path.realpath(local_file_path) + rel_path = os.path.relpath(canonical_target, canonical_root) + components = Path(rel_path).parts + + dir_fd_supported = os.open in os.supports_dir_fd + + if dir_fd_supported: + fd = None + try: + # Open the allowed root directory + fd = os.open(canonical_root, os.O_RDONLY | os.O_DIRECTORY) + + # Walk through the subdirectories resolving relative components + for comp in components[:-1]: + if comp == "." or not comp: + continue + next_fd = os.open( + comp, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd + ) + os.close(fd) + fd = next_fd + + # Open the final file component with O_NOFOLLOW + final_comp = components[-1] + file_flags = os.O_RDONLY | os.O_NOFOLLOW + file_fd = os.open(final_comp, file_flags, dir_fd=fd) + os.close(fd) + fd = None # fd is closed, only file_fd remains open + + # Verify that it is a regular file + stat_result = os.fstat(file_fd) + import stat + + if not stat.S_ISREG(stat_result.st_mode): + os.close(file_fd) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=ErrorResponse( + success=False, + error="Invalid file", + message="The target path is not a regular file.", + ).model_dump(), + ) + + with open(file_fd, "rb") as f: + image_bytes = f.read() + except Exception as e: + if fd is not None: + try: + os.close(fd) + except Exception: + pass + logger.error( + f"Failed to securely walk and open path {local_file_path}: {e}" + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ErrorResponse( + success=False, + error="Access Denied", + message="Cannot read the specified path.", + ).model_dump(), + ) + else: + # Fallback for platforms where dir_fd is not supported (Windows) + try: + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd_handle = os.open(local_file_path, flags) + + # Verify regular file using fstat on the opened fd + stat_result = os.fstat(fd_handle) + import stat + + if not stat.S_ISREG(stat_result.st_mode): + os.close(fd_handle) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=ErrorResponse( + success=False, + error="Invalid file", + message="The target path is not a regular file.", + ).model_dump(), + ) + + with open(fd_handle, "rb") as f: + image_bytes = f.read() + except Exception as e: + logger.error(f"Failed to securely open path {local_file_path}: {e}") + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ErrorResponse( + success=False, + error="Access Denied", + message="Cannot read the specified path.", + ).model_dump(), + ) return perform_face_search(image_bytes=image_bytes) diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index 4cd35ad5d..fe013e9c5 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -482,17 +482,36 @@ def test_face_search_safe_path_allowed(self, mock_perform, mock_folders): mock_folders.return_value = ["/allowed/folder"] mock_perform.return_value = {"success": True, "data": []} - from unittest.mock import mock_open + from unittest.mock import mock_open, MagicMock + import os + + mock_stat = MagicMock() + mock_stat.st_mode = 32768 # S_IFREG with patch("os.path.isfile", return_value=True), patch( "os.open", return_value=123 - ), patch("builtins.open", mock_open(read_data=b"fakeimagebytes")): + ) as mock_os_open, patch("os.fstat", return_value=mock_stat), patch( + "builtins.open", mock_open(read_data=b"fakeimagebytes") + ): response = client.post( "/face_clusters/face-search?input_type=path", json={"path": "/allowed/folder/family.jpg", "base64_data": ""}, ) assert response.status_code == 200 mock_perform.assert_called_once() + # Assert perform_face_search receives exactly b"fakeimagebytes" + called_kwargs = mock_perform.call_args.kwargs + assert called_kwargs.get("image_bytes") == b"fakeimagebytes" + # Assert os.open includes O_NOFOLLOW when supported + if hasattr(os, "O_NOFOLLOW"): + assert mock_os_open.called + nofollow_flag_used = False + for call in mock_os_open.call_args_list: + flags = call[0][1] if len(call[0]) > 1 else call[1].get("flags", 0) + if flags & os.O_NOFOLLOW: + nofollow_flag_used = True + break + assert nofollow_flag_used, "os.open was not called with O_NOFOLLOW flag" @patch("app.routes.face_clusters.perform_face_search") def test_face_search_base64_allowed(self, mock_perform): @@ -507,10 +526,12 @@ def test_face_search_base64_allowed(self, mock_perform): ) assert response.status_code == 200 mock_perform.assert_called_once() - # Verify that perform_face_search was called with image_bytes (not image_path) + # Decode the fixture and assert perform_face_search receives those exact bytes + import base64 + + expected_bytes = base64.b64decode(gif_b64.split(",")[-1]) called_kwargs = mock_perform.call_args.kwargs - assert "image_bytes" in called_kwargs - assert called_kwargs["image_bytes"] is not None + assert called_kwargs.get("image_bytes") == expected_bytes assert called_kwargs.get("image_path") is None