fix: resolve media file by directory listing to kill path-injection taint

Match the validated filename against iterdir() entries with == instead of
building a path from the user string, so no tainted value ever reaches a
filesystem sink. Satisfies CodeQL py/path-injection.

ha-relevant: no
This commit is contained in:
Pouzor
2026-07-07 14:35:44 +02:00
parent b907c4b05e
commit 7db57bba3b
+14 -16
View File
@@ -42,19 +42,21 @@ _NAME_RE = re.compile(r"[0-9a-f]{32}\.(png|jpg|webp)")
def _resolve_media_path(filename: str) -> Path: def _resolve_media_path(filename: str) -> Path:
"""Resolve `filename` inside the media dir, rejecting anything that escapes it. """Return the existing media file named `filename`, or raise 404.
`filename` is validated against `_NAME_RE` (no separators, no `..`) and the `filename` is validated against `_NAME_RE` (no separators, no `..`), then
resolved path is confirmed to sit directly under the resolved media dir. matched by name against the actual directory listing. The user string is
Raises 404 on any mismatch so we never leak whether a path exists elsewhere. only ever compared with `==` against trusted `iterdir()` entries — it never
builds a path — so a crafted value cannot escape the media dir.
""" """
if not _NAME_RE.fullmatch(filename): if not _NAME_RE.fullmatch(filename):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
base = settings.media_dir().resolve() base = settings.media_dir()
path = (base / filename).resolve() if base.is_dir():
if path.parent != base: for entry in base.iterdir():
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") if entry.name == filename and entry.is_file():
return path return entry
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
@router.post("/upload") @router.post("/upload")
@@ -89,14 +91,10 @@ async def upload_media(file: UploadFile, _user: str = Depends(get_current_user))
@router.get("/{filename}") @router.get("/{filename}")
async def get_media(filename: str) -> FileResponse: async def get_media(filename: str) -> FileResponse:
path = _resolve_media_path(filename) # _resolve_media_path returns only an existing file, else raises 404.
if not path.is_file(): return FileResponse(_resolve_media_path(filename))
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
return FileResponse(path)
@router.delete("/{filename}", status_code=status.HTTP_204_NO_CONTENT) @router.delete("/{filename}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_media(filename: str, _user: str = Depends(get_current_user)) -> None: async def delete_media(filename: str, _user: str = Depends(get_current_user)) -> None:
path = _resolve_media_path(filename) _resolve_media_path(filename).unlink()
if path.is_file():
path.unlink()