From e6bdd91a27b6af4babad02d3ab85831a0e7a90f7 Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Thu, 23 Jul 2026 10:20:28 +0000 Subject: [PATCH] fix(security): make personal path confinement explicit --- routes/personal_routes.py | 52 ++--- tests/test_personal_dir_symlink_escape.py | 177 +++++++++++++++--- tests/test_personal_remove_dir_confinement.py | 22 +-- 3 files changed, 187 insertions(+), 64 deletions(-) diff --git a/routes/personal_routes.py b/routes/personal_routes.py index a42615be7..7a4d0b74c 100644 --- a/routes/personal_routes.py +++ b/routes/personal_routes.py @@ -145,25 +145,6 @@ def _rag(): """Get the current RAG manager, retrying init if needed.""" return get_rag_manager() - def _resolve_allowed_personal_dir(directory: str) -> str: - """Resolve a user-supplied personal-docs path under the allowed root.""" - if not directory: - raise HTTPException(400, "Directory path is required") - - # realpath (not abspath) so a symlink inside PERSONAL_DIR that points - # outside it is resolved before the commonpath confinement check below; - # abspath only normalises `..` and would let such a symlink escape. - base_abs = os.path.realpath(PERSONAL_DIR) - candidate = directory if os.path.isabs(directory) else os.path.join(base_abs, directory) - resolved = os.path.realpath(candidate) - try: - in_base = os.path.commonpath([resolved, base_abs]) == base_abs - except ValueError: - in_base = False - if not in_base: - raise HTTPException(403, "Directory must be inside personal documents") - return resolved - @router.get("") def api_personal_list(owner: str = Depends(require_user), _admin: None = Depends(require_admin)): """Enhanced version that includes directories""" @@ -193,7 +174,20 @@ async def add_directory_to_rag( """ directory = directory_request.directory try: - directory = _resolve_allowed_personal_dir(directory) + if not directory: + raise HTTPException(400, "Directory path is required") + + # Keep normalization and the explicit prefix guard in this route: + # CodeQL carries the safe-access fact from startswith() to the + # filesystem sinks below only within this dataflow scope. + base_abs = os.path.realpath(PERSONAL_DIR) + candidate = directory if os.path.isabs(directory) else os.path.join(base_abs, directory) + directory = os.path.realpath(candidate) + base_prefix = base_abs if base_abs.endswith(os.sep) else base_abs + os.sep + if not directory.startswith(base_abs): + raise HTTPException(403, "Directory must be inside personal documents") + if directory != base_abs and not directory.startswith(base_prefix): + raise HTTPException(403, "Directory must be inside personal documents") # Security check - ensure directory exists and is accessible if not os.path.exists(directory): @@ -243,11 +237,19 @@ async def remove_directory_from_rag(directory: str = Query(...), owner: str = De JSON response confirming removal """ try: - # Confine to PERSONAL_DIR — parity with add_directory_to_rag (which - # resolves the path the same way). Without this, an arbitrary or - # `..`-escaping path is passed straight to - # personal_docs_manager.remove_directory / rag.remove_directory. - directory = _resolve_allowed_personal_dir(directory) + if not directory: + raise HTTPException(400, "Directory path is required") + + # Keep this guard local to the request source and removal sinks so + # CodeQL can prove the normalized path is confined. + base_abs = os.path.realpath(PERSONAL_DIR) + candidate = directory if os.path.isabs(directory) else os.path.join(base_abs, directory) + directory = os.path.realpath(candidate) + base_prefix = base_abs if base_abs.endswith(os.sep) else base_abs + os.sep + if not directory.startswith(base_abs): + raise HTTPException(403, "Directory must be inside personal documents") + if directory != base_abs and not directory.startswith(base_prefix): + raise HTTPException(403, "Directory must be inside personal documents") logger.info(f"Removing directory from RAG: {directory}") diff --git a/tests/test_personal_dir_symlink_escape.py b/tests/test_personal_dir_symlink_escape.py index 064e12c58..9e9d071c0 100644 --- a/tests/test_personal_dir_symlink_escape.py +++ b/tests/test_personal_dir_symlink_escape.py @@ -1,19 +1,15 @@ -"""Regression: _resolve_allowed_personal_dir must resolve symlinks (realpath) -when confining a path to PERSONAL_DIR. - -It used os.path.abspath, which normalises ``..`` but does NOT resolve symlinks, -so a symlink placed inside PERSONAL_DIR pointing outside it passes the -os.path.commonpath confinement check and lets index_personal_documents read -files outside the root. os.path.realpath resolves the symlink before the check. - -_resolve_allowed_personal_dir is a closure inside setup_personal_routes, so the -source-level test pins the fix and the behavioural test proves the underlying -confinement principle. -""" +"""Regression tests for personal-directory path confinement.""" import ast +import asyncio import os from pathlib import Path +import pytest +from fastapi import HTTPException + +from routes import personal_routes +from src.request_models import DirectoryRequest + SRC = Path(__file__).resolve().parent.parent / "routes" / "personal_routes.py" @@ -25,30 +21,155 @@ def _function_source(src_text, name): raise AssertionError(f"{name} not found in {SRC}") -def test_confinement_uses_realpath_not_abspath(): - body = _function_source(SRC.read_text(), "_resolve_allowed_personal_dir") +class _FakePersonalDocs: + index = [] + + def __init__(self): + self.added = [] + self.removed = [] + + def add_directory(self, directory, index=False): + self.added.append((directory, index)) + + def remove_directory(self, directory): + self.removed.append(directory) + + +class _FakeRAG: + def __init__(self): + self.indexed = [] + self.removed = [] + + def index_personal_documents(self, directory, owner=None): + self.indexed.append((directory, owner)) + return {"success": True, "indexed_count": 0, "failed_count": 0} + + def remove_directory(self, directory): + self.removed.append(directory) + + +def _endpoint(personal_docs, method, path): + router = personal_routes.setup_personal_routes(personal_docs, None, True) + for route in router.routes: + if ( + getattr(route, "path", "") == path + and method in getattr(route, "methods", set()) + ): + return route.endpoint + raise AssertionError(f"{method} {path} endpoint not found") + + +def _invoke(operation, directory, personal_docs): + if operation == "add": + endpoint = _endpoint(personal_docs, "POST", "/api/personal/add_directory") + return asyncio.run( + endpoint( + request=object(), + directory_request=DirectoryRequest(directory=directory), + owner="alice", + _admin=None, + ) + ) + + endpoint = _endpoint(personal_docs, "DELETE", "/api/personal/remove_directory") + return asyncio.run(endpoint(directory=directory, owner="alice", _admin=None)) + + +@pytest.mark.parametrize( + "route_name", + ["add_directory_to_rag", "remove_directory_from_rag"], +) +def test_confinement_is_inline_and_codeql_visible(route_name): + body = _function_source(SRC.read_text(), route_name) assert "os.path.realpath" in body, ( - "_resolve_allowed_personal_dir must use os.path.realpath so a symlink " - "inside PERSONAL_DIR cannot escape the confinement check" + f"{route_name} must resolve symlinks before checking confinement" ) - assert "os.path.abspath" not in body, ( - "os.path.abspath does not resolve symlinks; the confinement check must " - "not rely on it" + assert "os.path.abspath" not in body + assert "os.path.commonpath" not in body + assert "os.path.normcase" not in body + assert "if not directory.startswith(base_abs):" in body, ( + "the CodeQL-recognized safe-access guard must run on every accepted path" ) + assert body.index("if not directory.startswith(base_abs):") < body.index( + "if directory != base_abs" + ), "the broad analyzer guard must run before the separator-safe boundary check" + + +@pytest.mark.parametrize("operation", ["add", "remove"]) +@pytest.mark.parametrize("directory_kind", ["root", "relative", "absolute"]) +def test_confinement_accepts_root_and_descendants( + operation, directory_kind, tmp_path, monkeypatch +): + base = tmp_path / "PersonalDocs" + nested = base / "nested" + nested.mkdir(parents=True) + monkeypatch.setattr(personal_routes, "PERSONAL_DIR", str(base)) + rag = _FakeRAG() + monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag) + docs = _FakePersonalDocs() + + supplied = { + "root": ".", + "relative": "nested", + "absolute": str(nested), + }[directory_kind] + expected = os.path.realpath(base if directory_kind == "root" else nested) + result = _invoke(operation, supplied, docs) + + assert result["success"] is True + assert result["directory"] == expected + if operation == "add": + assert docs.added == [(expected, False)] + assert rag.indexed == [(expected, "alice")] + else: + assert docs.removed == [expected] + assert rag.removed == [expected] -def test_realpath_catches_symlink_escape(tmp_path): - # The principle the fix relies on: abspath keeps the symlink path inside the - # base (confinement fooled); realpath resolves it outside (confinement holds). +@pytest.mark.parametrize("operation", ["add", "remove"]) +def test_confinement_rejects_parent_and_sibling_prefix( + operation, tmp_path, monkeypatch +): + base = tmp_path / "personal" + base.mkdir() + sibling = tmp_path / "personal-backup" + sibling.mkdir() + monkeypatch.setattr(personal_routes, "PERSONAL_DIR", str(base)) + rag = _FakeRAG() + monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag) + + for directory in ("..", str(sibling)): + docs = _FakePersonalDocs() + with pytest.raises(HTTPException) as exc: + _invoke(operation, directory, docs) + assert exc.value.status_code == 403 + assert docs.added == [] + assert docs.removed == [] + assert rag.indexed == [] + assert rag.removed == [] + + +@pytest.mark.parametrize("operation", ["add", "remove"]) +def test_confinement_rejects_symlink_escape(operation, tmp_path, monkeypatch): base = tmp_path / "personal" base.mkdir() outside = tmp_path / "outside" outside.mkdir() link = base / "escape" - os.symlink(outside, link) + try: + os.symlink(outside, link) + except (AttributeError, NotImplementedError, OSError) as exc: + pytest.skip(f"symlinks unavailable: {exc}") + monkeypatch.setattr(personal_routes, "PERSONAL_DIR", str(base)) + rag = _FakeRAG() + monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag) + docs = _FakePersonalDocs() + + with pytest.raises(HTTPException) as exc: + _invoke(operation, str(link), docs) - base_abs = os.path.realpath(base) # base itself may live under a symlinked tmp - # abspath: the symlink still looks inside base -> escape not detected - assert os.path.commonpath([os.path.abspath(base / "escape"), os.path.abspath(base)]) == os.path.abspath(base) - # realpath: the symlink resolves to `outside` -> escape detected - assert os.path.commonpath([os.path.realpath(link), base_abs]) != base_abs + assert exc.value.status_code == 403 + assert docs.added == [] + assert docs.removed == [] + assert rag.indexed == [] + assert rag.removed == [] diff --git a/tests/test_personal_remove_dir_confinement.py b/tests/test_personal_remove_dir_confinement.py index a869d7bf9..a61131e14 100644 --- a/tests/test_personal_remove_dir_confinement.py +++ b/tests/test_personal_remove_dir_confinement.py @@ -3,11 +3,10 @@ DELETE /api/personal/remove_directory took a raw ``directory`` query parameter and passed it straight to ``personal_docs_manager.remove_directory`` / ``rag.remove_directory`` with no containment check — unlike add_directory_to_rag, -which resolves the path via ``_resolve_allowed_personal_dir`` first. This pins -the parity fix. +which normalizes and confines the path first. This pins the parity fix. -``_resolve_allowed_personal_dir`` is a closure inside ``setup_personal_routes``, -so this is a source-level test, matching test_personal_dir_symlink_escape.py. +The source-level checks complement the direct behavioral tests in +test_personal_dir_symlink_escape.py. """ import ast from pathlib import Path @@ -25,19 +24,20 @@ def _function_source(src_text: str, name: str) -> str: def test_remove_directory_confines_path(): body = _function_source(SRC.read_text(), "remove_directory_from_rag") - assert "_resolve_allowed_personal_dir(" in body, ( - "remove_directory_from_rag must call _resolve_allowed_personal_dir to " - "confine the user-supplied directory to PERSONAL_DIR (parity with " - "add_directory_to_rag)" + assert "os.path.realpath" in body, ( + "remove_directory_from_rag must normalize the user-supplied directory" + ) + assert "if not directory.startswith(base_abs):" in body, ( + "remove_directory_from_rag must use a CodeQL-visible confinement guard" ) def test_confinement_runs_before_removal_sinks(): """The confinement must happen before the path reaches either removal sink.""" body = _function_source(SRC.read_text(), "remove_directory_from_rag") - resolve_idx = body.index("_resolve_allowed_personal_dir(") + guard_idx = body.index("if not directory.startswith(base_abs):") for sink in ("personal_docs_manager.remove_directory(", "rag.remove_directory("): assert sink in body, f"expected sink {sink} in remove_directory_from_rag" - assert body.index(sink) > resolve_idx, ( - f"{sink} runs before _resolve_allowed_personal_dir — path not confined" + assert body.index(sink) > guard_idx, ( + f"{sink} runs before the confinement guard" )