fix(security): make personal path confinement CodeQL-visible

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}")
 

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 == []

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"
         )