fix: path traversal vulnerabilities in migration image imports and media routes (#7474)

hay-kot · web-flow · commit bd296c3eafb8 · 2026-04-16T03:24:50.000Z
diff --git a/mealie/routes/admin/admin_backups.py b/mealie/routes/admin/admin_backups.py
@@ -19,8 +19,12 @@
 
 @controller(router)
 class AdminBackupController(BaseAdminController):
-    def _backup_path(self, name) -> Path:
-        return get_app_dirs().BACKUP_DIR / name
+    def _backup_path(self, name: str) -> Path:
+        backup_dir = get_app_dirs().BACKUP_DIR
+        candidate = (backup_dir / name).resolve()
+        if not candidate.is_relative_to(backup_dir.resolve()):
+            raise HTTPException(status.HTTP_400_BAD_REQUEST)
+        return candidate
 
     @router.get("", response_model=AllBackups)
     def get_all(self):
@@ -86,7 +90,7 @@ def upload_one(self, archive: UploadFile = File(...)):
         app_dirs = get_app_dirs()
         dest = app_dirs.BACKUP_DIR.joinpath(f"{name}.zip")
 
-        if dest.absolute().parent != app_dirs.BACKUP_DIR:
+        if dest.resolve().parent != app_dirs.BACKUP_DIR.resolve():
             raise HTTPException(status.HTTP_400_BAD_REQUEST)
 
         with dest.open("wb") as buffer:
diff --git a/mealie/routes/admin/admin_debug.py b/mealie/routes/admin/admin_debug.py
@@ -1,5 +1,6 @@
 import os
 import shutil
+from pathlib import Path
 
 from fastapi import APIRouter, File, UploadFile
 
@@ -25,9 +26,12 @@ async def debug_openai(self, image: UploadFile | None = File(None)):
 
         with get_temporary_path() as temp_path:
             if image:
-                with temp_path.joinpath(image.filename).open("wb") as buffer:
+                if not image.filename:
+                    return DebugResponse(success=False, response="Invalid image filename")
+                safe_filename = Path(image.filename).name
+                local_image_path = temp_path.joinpath(safe_filename)
+                with local_image_path.open("wb") as buffer:
                     shutil.copyfileobj(image.file, buffer)
-                local_image_path = temp_path.joinpath(image.filename)
                 local_images = [OpenAILocalImage(filename=os.path.basename(local_image_path), path=local_image_path)]
             else:
                 local_images = None
diff --git a/mealie/routes/media/media_recipe.py b/mealie/routes/media/media_recipe.py
@@ -17,7 +17,7 @@ class ImageType(StrEnum):
 
 
 @router.get("/{recipe_id}/images/{file_name}")
-async def get_recipe_img(recipe_id: str, file_name: ImageType = ImageType.original):
+async def get_recipe_img(recipe_id: UUID4, file_name: ImageType = ImageType.original):
     """
     Takes in a recipe id, returns the static image. This route is proxied in the docker image
     and should not hit the API in production
@@ -32,7 +32,7 @@ async def get_recipe_img(recipe_id: str, file_name: ImageType = ImageType.origin
 
 @router.get("/{recipe_id}/images/timeline/{timeline_event_id}/{file_name}")
 async def get_recipe_timeline_event_img(
-    recipe_id: str, timeline_event_id: str, file_name: ImageType = ImageType.original
+    recipe_id: UUID4, timeline_event_id: UUID4, file_name: ImageType = ImageType.original
 ):
     """
     Takes in a recipe id and event timeline id, returns the static image. This route is proxied in the docker image
@@ -51,7 +51,11 @@ async def get_recipe_timeline_event_img(
 @router.get("/{recipe_id}/assets/{file_name}")
 async def get_recipe_asset(recipe_id: UUID4, file_name: str):
     """Returns a recipe asset"""
-    file = Recipe.directory_from_id(recipe_id).joinpath("assets", file_name)
+    asset_dir = Recipe.directory_from_id(recipe_id).joinpath("assets")
+    file = asset_dir.joinpath(file_name).resolve()
+
+    if not file.is_relative_to(asset_dir.resolve()):
+        raise HTTPException(status.HTTP_400_BAD_REQUEST)
 
     if file.exists():
         return FileResponse(file)
diff --git a/mealie/routes/media/media_user.py b/mealie/routes/media/media_user.py
@@ -11,7 +11,11 @@
 async def get_user_image(user_id: UUID4, file_name: str):
     """Takes in a recipe slug, returns the static image. This route is proxied in the docker image
     and should not hit the API in production"""
-    recipe_image = PrivateUser.get_directory(user_id) / file_name
+    user_dir = PrivateUser.get_directory(user_id)
+    recipe_image = (user_dir / file_name).resolve()
+
+    if not recipe_image.is_relative_to(user_dir.resolve()):
+        raise HTTPException(status.HTTP_400_BAD_REQUEST)
 
     if recipe_image.exists():
         return FileResponse(recipe_image, media_type="image/webp")
diff --git a/mealie/services/migrations/_migration_base.py b/mealie/services/migrations/_migration_base.py
@@ -272,8 +272,8 @@ def clean_recipe_dictionary(self, recipe_dict: dict) -> Recipe:
         recipe = cleaner.clean(recipe_dict, self.translator, url=recipe_dict.get("org_url", None))
         return recipe
 
-    def import_image(self, slug: str, src: str | Path, recipe_id: UUID4):
+    def import_image(self, slug: str, src: str | Path, recipe_id: UUID4, extraction_root: Path | None = None):
         try:
-            import_image(src, recipe_id)
+            import_image(src, recipe_id, extraction_root=extraction_root)
         except UnidentifiedImageError as e:
             self.logger.error(f"Failed to import image for {slug}: {e}")
diff --git a/mealie/services/migrations/chowdown.py b/mealie/services/migrations/chowdown.py
@@ -4,7 +4,7 @@
 
 from ._migration_base import BaseMigrator
 from .utils.migration_alias import MigrationAlias
-from .utils.migration_helpers import MigrationReaders, split_by_comma
+from .utils.migration_helpers import MigrationReaders, safe_local_path, split_by_comma
 
 
 class ChowdownMigrator(BaseMigrator):
@@ -60,8 +60,10 @@ def _migrate(self) -> None:
                             continue
 
                         if r.image:
-                            cd_image = image_dir.joinpath(r.image)
+                            cd_image = safe_local_path(image_dir.joinpath(r.image), image_dir)
+                        else:
+                            cd_image = None
                     except StopIteration:
                         continue
                     if cd_image:
-                        self.import_image(slug, cd_image, recipe_id)
+                        self.import_image(slug, cd_image, recipe_id, extraction_root=image_dir)
diff --git a/mealie/services/migrations/cookn.py b/mealie/services/migrations/cookn.py
@@ -11,7 +11,7 @@
 from mealie.services.parser_services.parser_utils.string_utils import extract_quantity_from_string
 
 from ._migration_base import BaseMigrator
-from .utils.migration_helpers import format_time
+from .utils.migration_helpers import format_time, safe_local_path
 
 
 class DSVParser:
@@ -157,15 +157,21 @@ def _parse_media(self, _cookbook_id: str, _chapter_id: str, _recipe_id: str, db:
                 if _media_type != "":
                     # Determine file extension based on media type
                     _extension = _media_type.split("/")[-1]
-                    _old_image_path = os.path.join(db.directory, str(_media_id))
-                    new_image_path = f"{_old_image_path}.{_extension}"
+                    _old_image_path = Path(db.directory) / str(_media_id)
+                    new_image_path = _old_image_path.with_suffix(f".{_extension}")
+                    if safe_local_path(_old_image_path, db.directory) is None:
+                        return None
+                    if safe_local_path(new_image_path, db.directory) is None:
+                        return None
                     # Rename the file if it exists and has no extension
-                    if os.path.exists(_old_image_path) and not os.path.exists(new_image_path):
+                    if _old_image_path.exists() and not new_image_path.exists():
                         os.rename(_old_image_path, new_image_path)
-                    if Path(new_image_path).exists():
-                        return new_image_path
+                    if new_image_path.exists():
+                        return str(new_image_path)
             else:
-                return os.path.join(db.directory, str(_media_id))
+                candidate = Path(db.directory) / str(_media_id)
+                if safe_local_path(candidate, db.directory) is not None:
+                    return str(candidate)
         return None
 
     def _parse_ingredients(self, _recipe_id: str, db: DSVParser) -> list[RecipeIngredient]:
@@ -388,14 +394,14 @@ def _process_cookbook(self, path: Path) -> None:
                 recipe = recipe_lookup.get(slug)
                 if recipe:
                     if recipe.image:
-                        self.import_image(slug, recipe.image, recipe_id)
+                        self.import_image(slug, recipe.image, recipe_id, extraction_root=db.directory)
                 else:
                     index_len = len(slug.split("-")[-1])
                     recipe = recipe_lookup.get(slug[: -(index_len + 1)])
                     if recipe:
                         self.logger.warning("Duplicate recipe (%s) found! Saved as copy...", recipe.name)
                         if recipe.image:
-                            self.import_image(slug, recipe.image, recipe_id)
+                            self.import_image(slug, recipe.image, recipe_id, extraction_root=db.directory)
                     else:
                         self.logger.warning("Failed to lookup recipe! (%s)", slug)
 
diff --git a/mealie/services/migrations/copymethat.py b/mealie/services/migrations/copymethat.py
@@ -9,7 +9,7 @@
 
 from ._migration_base import BaseMigrator
 from .utils.migration_alias import MigrationAlias
-from .utils.migration_helpers import import_image
+from .utils.migration_helpers import import_image, safe_local_path
 
 
 def parse_recipe_tags(tags: list) -> list[str]:
@@ -52,7 +52,9 @@ def _process_recipe_document(self, source_dir: Path, soup: BeautifulSoup) -> dic
             # the recipe image tag has no id, so we parse it directly
             if tag.name == "img" and "recipeImage" in tag.get("class", []):
                 if image_path := tag.get("src"):
-                    recipe_dict["image"] = str(source_dir.joinpath(image_path))
+                    safe = safe_local_path(source_dir.joinpath(image_path), source_dir)
+                    if safe is not None:
+                        recipe_dict["image"] = str(safe)
 
                 continue
 
@@ -120,4 +122,4 @@ def _migrate(self) -> None:
                     except StopIteration:
                         continue
 
-                    import_image(r.image, recipe_id)
+                    import_image(r.image, recipe_id, extraction_root=source_dir)
diff --git a/mealie/services/migrations/nextcloud.py b/mealie/services/migrations/nextcloud.py
@@ -97,4 +97,4 @@ def _migrate(self) -> None:
                 if status:
                     nc_dir = nextcloud_dirs[slug]
                     if nc_dir.image:
-                        self.import_image(slug, nc_dir.image, recipe_id)
+                        self.import_image(slug, nc_dir.image, recipe_id, extraction_root=base_dir)
diff --git a/mealie/services/migrations/paprika.py b/mealie/services/migrations/paprika.py
@@ -84,6 +84,6 @@ def _migrate(self) -> None:
                     temp_file.write(image.read())
                     temp_file.flush()
                     path = Path(temp_file.name)
-                    self.import_image(slug, path, recipe_id)
+                    self.import_image(slug, path, recipe_id, extraction_root=path.parent)
             except Exception as e:
                 self.logger.error(f"Failed to import image for {slug}: {e}")
diff --git a/mealie/services/migrations/recipekeeper.py b/mealie/services/migrations/recipekeeper.py
@@ -8,7 +8,7 @@
 
 from ._migration_base import BaseMigrator
 from .utils.migration_alias import MigrationAlias
-from .utils.migration_helpers import parse_iso8601_duration
+from .utils.migration_helpers import parse_iso8601_duration, safe_local_path
 
 
 def clean_instructions(instructions: list[str]) -> list[str]:
@@ -30,7 +30,9 @@ def parse_recipe_div(recipe, image_path):
         elif item.name == "div":
             meta[item["itemprop"]] = list(item.stripped_strings)
         elif item.name == "img":
-            meta[item["itemprop"]] = str(image_path / item["src"])
+            safe = safe_local_path(image_path / item["src"], image_path)
+            if safe is not None:
+                meta[item["itemprop"]] = str(safe)
         else:
             meta[item["itemprop"]] = item.string
     # merge nutrition keys into their own dict.
@@ -107,4 +109,4 @@ def _migrate(self) -> None:
                     except StopIteration:
                         continue
 
-                    self.import_image(slug, recipe.image, recipe_id)
+                    self.import_image(slug, recipe.image, recipe_id, extraction_root=source_dir)
diff --git a/mealie/services/migrations/tandoor.py b/mealie/services/migrations/tandoor.py
@@ -132,4 +132,4 @@ def _migrate(self) -> None:
                     except StopIteration:
                         continue
 
-                    self.import_image(slug, r.image, recipe_id)
+                    self.import_image(slug, r.image, recipe_id, extraction_root=source_dir)
diff --git a/mealie/services/migrations/utils/migration_helpers.py b/mealie/services/migrations/utils/migration_helpers.py
@@ -8,6 +8,7 @@
 from PIL import UnidentifiedImageError
 from pydantic import UUID4
 
+from mealie.core import root_logger
 from mealie.services.recipe.recipe_data_service import RecipeDataService
 
 
@@ -100,16 +101,45 @@ def glob_walker(directory: Path, glob_str: str, return_parent=True) -> list[Path
     return matches
 
 
-def import_image(src: str | Path, recipe_id: UUID4):
-    """Read the successful migrations attribute and for each import the image
-    appropriately into the image directory. Minification is done in mass
-    after the migration occurs.
+def safe_local_path(candidate: str | Path, root: Path) -> Path | None:
+    """
+    Returns the resolved path only if it is safely contained within root.
+
+    Returns ``None`` for any path that would escape the root directory,
+    including ``../../`` traversal sequences and absolute paths outside root.
+    Symlinks are followed by ``resolve()``, so a symlink pointing outside root
+    is also rejected.
+    """
+    try:
+        # OSError: symlink resolution failure; ValueError: null bytes on some platforms
+        resolved = Path(candidate).resolve()
+        if resolved.is_relative_to(root.resolve()):
+            return resolved
+    except (OSError, ValueError):
+        pass
+    return None
+
+
+def import_image(src: str | Path, recipe_id: UUID4, extraction_root: Path | None = None):
+    """Import a local image file into the recipe image directory.
+
     May raise an UnidentifiedImageError if the file is not a recognised format.
+
+    If extraction_root is provided, the src path must be contained within it.
+    Paths that escape the extraction_root are silently rejected to prevent
+    arbitrary local file reads via archive-controlled image paths.
     """
 
     if isinstance(src, str):
         src = Path(src)
 
+    if extraction_root is not None:
+        if safe_local_path(src, extraction_root) is None:
+            root_logger.get_logger().warning(
+                "Rejected image path outside extraction root: %s (root: %s)", src, extraction_root
+            )
+            return
+
     if not src.exists():
         return
 
diff --git a/mealie/services/openai/openai.py b/mealie/services/openai/openai.py
@@ -129,19 +129,24 @@ def _get_prompt_file_candidates(self, name: str) -> list[Path]:
         """
         tree = name.split(".")
         relative_path = Path(*tree[:-1], tree[-1] + ".txt")
-        default_prompt_file = Path(self.PROMPTS_DIR, relative_path)
+
+        default_prompt_file = (self.PROMPTS_DIR / relative_path).resolve()
+        if not default_prompt_file.is_relative_to(self.PROMPTS_DIR.resolve()):
+            raise ValueError(f"Invalid prompt name '{name}': resolves outside prompts directory")
 
         try:
             # Only include custom files if the custom_dir is configured, is a directory, and the prompt file exists
-            custom_dir = Path(self.custom_prompt_dir) if self.custom_prompt_dir else None
+            custom_dir = Path(self.custom_prompt_dir).resolve() if self.custom_prompt_dir else None
             if custom_dir and not custom_dir.is_dir():
                 custom_dir = None
         except Exception:
             custom_dir = None
 
         if custom_dir:
-            custom_prompt_file = Path(custom_dir, relative_path)
-            if custom_prompt_file.exists():
+            custom_prompt_file = (custom_dir / relative_path).resolve()
+            if not custom_prompt_file.is_relative_to(custom_dir):
+                logger.warning(f"Custom prompt file resolves outside custom dir, skipping: {custom_prompt_file}")
+            elif custom_prompt_file.exists():
                 logger.debug(f"Found valid custom prompt file: {custom_prompt_file}")
                 return [custom_prompt_file, default_prompt_file]
             else:
diff --git a/mealie/services/recipe/recipe_data_service.py b/mealie/services/recipe/recipe_data_service.py
@@ -99,7 +99,12 @@ def write_image(self, file_data: bytes | Path, extension: str, image_dir: Path |
             with open(image_path, "ab") as f:
                 shutil.copyfileobj(file_data, f)
 
-        self.minifier.minify(image_path)
+        try:
+            self.minifier.minify(image_path)
+        except Exception:
+            # Remove the partially-written file so corrupt images don't persist on disk.
+            image_path.unlink(missing_ok=True)
+            raise
 
         return image_path
 
diff --git a/mealie/services/recipe/recipe_service.py b/mealie/services/recipe/recipe_service.py
@@ -325,9 +325,11 @@ async def create_from_images(self, images: list[UploadFile], translate_language:
         with get_temporary_path() as temp_path:
             local_images: list[Path] = []
             for image in images:
-                with temp_path.joinpath(image.filename).open("wb") as buffer:
+                safe_filename = Path(image.filename).name
+                image_path = temp_path.joinpath(safe_filename)
+                with image_path.open("wb") as buffer:
                     shutil.copyfileobj(image.file, buffer)
-                local_images.append(temp_path.joinpath(image.filename))
+                local_images.append(image_path)
 
             recipe_data = await openai_recipe_service.build_recipe_from_images(
                 local_images, translate_language=translate_language
