fix(backend): prevent path traversal in face search endpoint

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,
+    ) -> 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:
+            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 '<bytes>'}")
             return None
 
         boxes, scores, class_ids = self.yolo_detector(img)

backend/app/routes/face_clusters.py

@@ -2,8 +2,7 @@
 from binascii import Error as Base64Error
 import base64
 from concurrent.futures import CancelledError, Future, ProcessPoolExecutor
-from typing import Annotated
-import uuid
+from typing import Annotated, Optional
 import os
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from app.database.face_clusters import (
@@ -37,11 +36,40 @@
 )
 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 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()
+    except Exception:
+        allowed_folders = []
+
+    temp_dir = os.path.realpath("temp_uploads")
+    allowed_folders.append(temp_dir)
+
+    for folder in allowed_folders:
+        real_folder = os.path.realpath(folder)
+        try:
+            if os.path.commonpath([real_folder, real_target]) == real_folder:
+                return real_folder
+        except ValueError:
+            continue
+    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:
     """
     Report a rescore that died in the worker.
@@ -267,8 +295,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
 
@@ -281,6 +307,16 @@ def face_tagging(
                     message="image path is required.",
                 ).model_dump(),
             )
+        allowed_root = get_safe_root(local_file_path)
+        if not allowed_root:
+            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(),
+            )
         if not os.path.isfile(local_file_path):
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
@@ -290,7 +326,110 @@ def face_tagging(
                     message="The provided path is not a valid file",
                 ).model_dump(),
             )
-        image_path = payload.path
+
+        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)
 
     elif input_type == InputType.base64:
         base64_data = payload.base64_data
@@ -326,29 +465,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(

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,