|
12 | 12 |
|
13 | 13 | import logging |
14 | 14 | import mimetypes |
15 | | -import os |
16 | 15 | from pathlib import Path |
17 | 16 |
|
18 | 17 | from fastapi import APIRouter, HTTPException, Query |
|
61 | 60 | } |
62 | 61 |
|
63 | 62 |
|
64 | | -def _validate_media_path(*, path: str, allowed_root: str) -> str: |
| 63 | +def _validate_media_path(*, path: str, allowed_root: Path) -> Path: |
65 | 64 | """ |
66 | 65 | Validate and sanitize a user-provided file path against an allowed root directory. |
67 | 66 |
|
68 | | - Uses ``os.path.realpath`` to resolve symlinks and ``..`` components, then |
69 | | - verifies the canonical path starts with the allowed root prefix. This is |
70 | | - the standard sanitization pattern recognized by static analysis tools |
71 | | - (e.g. CodeQL ``py/path-injection``). |
| 67 | + Uses ``Path.resolve()`` to resolve symlinks and ``..`` components, then |
| 68 | + verifies the canonical path is under the allowed root. This is the standard |
| 69 | + sanitization pattern recognized by static analysis tools (e.g. CodeQL |
| 70 | + ``py/path-injection``). |
72 | 71 |
|
73 | 72 | Args: |
74 | 73 | path: The user-provided file path to validate. |
75 | | - allowed_root: The canonical (``realpath``-resolved) allowed root directory. |
| 74 | + allowed_root: The canonical (``resolve``-d) allowed root directory. |
76 | 75 |
|
77 | 76 | Returns: |
78 | 77 | The canonical, validated file path. |
79 | 78 |
|
80 | 79 | Raises: |
81 | 80 | HTTPException 403: If the path fails any validation check. |
82 | 81 | """ |
83 | | - real_path = os.path.realpath(path) |
84 | | - allowed_prefix = allowed_root + os.sep |
| 82 | + real_path = Path(path).resolve(strict=False) |
85 | 83 |
|
86 | | - if not real_path.startswith(allowed_prefix): |
87 | | - raise HTTPException(status_code=403, detail="Access denied: path is outside the allowed results directory.") |
| 84 | + try: |
| 85 | + relative_parts = real_path.relative_to(allowed_root).parts |
| 86 | + except ValueError as exc: |
| 87 | + raise HTTPException( |
| 88 | + status_code=403, detail="Access denied: path is outside the allowed results directory." |
| 89 | + ) from exc |
88 | 90 |
|
89 | 91 | # Restrict to known media subdirectories (e.g. prompt-memory-entries/) |
90 | | - relative_parts = Path(os.path.relpath(real_path, allowed_root)).parts |
91 | 92 | if not relative_parts or relative_parts[0] not in _ALLOWED_SUBDIRECTORIES: |
92 | 93 | raise HTTPException(status_code=403, detail="Access denied: path is not in a media subdirectory.") |
93 | 94 |
|
94 | 95 | # Only allow known media file extensions |
95 | | - _, ext = os.path.splitext(real_path) |
96 | | - if ext.lower() not in _ALLOWED_EXTENSIONS: |
| 96 | + if real_path.suffix.lower() not in _ALLOWED_EXTENSIONS: |
97 | 97 | raise HTTPException(status_code=403, detail="Access denied: file type is not allowed.") |
98 | 98 |
|
99 | 99 | return real_path |
@@ -125,13 +125,13 @@ async def serve_media_async( |
125 | 125 | memory = CentralMemory.get_memory_instance() |
126 | 126 | if not memory.results_path: |
127 | 127 | raise HTTPException(status_code=500, detail="Memory results_path is not configured.") |
128 | | - allowed_root = os.path.realpath(memory.results_path) |
| 128 | + allowed_root = Path(memory.results_path).resolve(strict=False) |
129 | 129 | except Exception as exc: |
130 | 130 | raise HTTPException(status_code=500, detail="Memory not initialized; cannot determine results path.") from exc |
131 | 131 |
|
132 | 132 | validated_path = _validate_media_path(path=path, allowed_root=allowed_root) |
133 | 133 |
|
134 | | - if not os.path.isfile(validated_path): |
| 134 | + if not validated_path.is_file(): |
135 | 135 | raise HTTPException(status_code=404, detail="File not found.") |
136 | 136 |
|
137 | 137 | mime_type, _ = mimetypes.guess_type(validated_path) |
|
0 commit comments