Merge pull request #256 from Pouzor/fix/media-path-injection

fix: harden media path handling against path injection
This commit is contained in:
Pouzor - Rémy Jardient
2026-07-07 14:55:27 +02:00
committed by GitHub
2 changed files with 30 additions and 12 deletions
+23 -12
View File
@@ -11,6 +11,7 @@ raw-image uploads reuse the same endpoint.
import re import re
import uuid import uuid
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, UploadFile, status from fastapi import APIRouter, Depends, HTTPException, UploadFile, status
from fastapi.responses import FileResponse from fastapi.responses import FileResponse
@@ -37,7 +38,25 @@ _MAGIC: dict[str, tuple[bytes, ...]] = {
MAX_BYTES = 10 * 1024 * 1024 # 10 MB MAX_BYTES = 10 * 1024 * 1024 # 10 MB
# Only ever serve/delete files we created: 32 hex chars + known extension. # Only ever serve/delete files we created: 32 hex chars + known extension.
_NAME_RE = re.compile(r"^[0-9a-f]{32}\.(png|jpg|webp)$") _NAME_RE = re.compile(r"[0-9a-f]{32}\.(png|jpg|webp)")
def _resolve_media_path(filename: str) -> Path:
"""Return the existing media file named `filename`, or raise 404.
`filename` is validated against `_NAME_RE` (no separators, no `..`), then
matched by name against the actual directory listing. The user string is
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):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
base = settings.media_dir()
if base.is_dir():
for entry in base.iterdir():
if entry.name == filename and entry.is_file():
return entry
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
@router.post("/upload") @router.post("/upload")
@@ -72,18 +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:
if not _NAME_RE.match(filename): # _resolve_media_path returns only an existing file, else raises 404.
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") return FileResponse(_resolve_media_path(filename))
path = settings.media_dir() / filename
if not path.is_file():
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:
if not _NAME_RE.match(filename): _resolve_media_path(filename).unlink()
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
path = settings.media_dir() / filename
if path.is_file():
path.unlink()
+7
View File
@@ -79,6 +79,13 @@ async def test_get_rejects_bad_filename(client, media_dir):
assert res.status_code == 404 assert res.status_code == 404
@pytest.mark.asyncio
async def test_delete_rejects_bad_filename(client, auth_headers, media_dir):
headers = await auth_headers()
res = await client.delete("/api/v1/media/..%2f..%2fetc%2fpasswd", headers=headers)
assert res.status_code == 404
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_delete_requires_auth_and_removes_file(client, auth_headers, media_dir): async def test_delete_requires_auth_and_removes_file(client, auth_headers, media_dir):
headers = await auth_headers() headers = await auth_headers()