fix(security): validate user-influenced file paths (py/path-injection) (#2252)

## Summary

CodeQL flagged 58 open `py/path-injection` alerts. Three were real gaps,
now fixed; the other 41 were false positives (paths already fully
validated upstream by an existing sanitizer CodeQL's dataflow engine
doesn't recognize) — dismissed individually with a per-alert
justification.

- **Hub agent install directory traversal**:
`agent_install_dir()`/`_backup_dir()` joined `agent_id` onto
`~/.gaia/agents/` for `rmtree`/`move`/write with no shape check. An id
like `../../evil` (reachable via `POST /api/agents/install` or a corrupt
sentinel) could point install/uninstall/rollback/configure at an
arbitrary directory. Now rejected with `InstallError` unless the id is a
single safe path component.
- **`safe_open_document` symlink-directory escape**: the home-directory
containment check used a lexical `os.path.abspath` comparison, which an
intermediate symlinked directory *inside* home (pointing outside) could
bypass — `O_NOFOLLOW` only blocks a symlinked *final* path component.
Added an `os.path.realpath`-based physical containment check.
- **EMR `compute_file_hash()` missing its own guard**: three call sites
in `gaia_agent_emr/agent.py` omitted the existing `allowed_dir`
parameter, so the hashing helper had no containment check at all if
reached with an untrusted path.

### Files fixed (with traversal tests)
- `src/gaia/hub/installer.py`, `src/gaia/hub/lifecycle.py`,
`src/gaia/ui/routers/hub.py` — new `_require_safe_agent_id()` guard +
sentinel `executable` field validation. Tests:
`tests/unit/test_hub_installer.py::TestAgentIdPathSafety` (traversal,
absolute path, null byte, oversized id, non-string id, sentinel with
unsafe executable).
- `src/gaia/ui/utils.py` — realpath containment check in
`safe_open_document()`. Test:
`tests/unit/chat/ui/test_toctou.py::test_safe_open_rejects_symlinked_directory_escape`.
- `hub/agents/python/emr/gaia_agent_emr/agent.py` — pass `allowed_dir`
at all three `compute_file_hash()` call sites. Tests:
`tests/unit/test_file_watcher.py::TestFileHashUtilities` (traversal,
absolute escape, contained-path-still-works).

### Alerts dismissed as false positive (41, listed by number — each has
a per-alert justification in its dismissal comment)
- `documents.py` upload-path lock key + `index_folder()`: 231, 184–188
(path already gated by `safe_open_document`/`ensure_within_home`
downstream)
- `files.py` browse/search/preview/image endpoints: 189–206, 244–248
(all gated by `ensure_within_home()`, or the search scan roots are
hardcoded to home subdirs with the user param only a filename substring
filter)
- `server.py` SPA static asset serving: 341, 342 (gated by
`sanitize_static_path()`, a real containment-checking sanitizer)
- `utils.py` sanitizer return: 209 (the sanitizer's own
already-validated return value)
- `file_watcher.py` `FileWatcher.__init__`: 315 (no caller passes
remote/HTTP-derived input)
- EMR dashboard/agent: 328–335 (watch_dir is set only via a
fully-validated setter — regex allowlist, normpath/abspath, realpath
symlink check, home-dir prefix, sensitive-dir denylist — before any of
these lines run)

## Test plan
- [ ] `PYTHONPATH=src python -m pytest tests/unit/test_hub_installer.py
tests/unit/test_hub_lifecycle.py tests/unit/test_hub_router.py
tests/unit/test_hub_security.py tests/unit/chat/ui/test_toctou.py
tests/unit/test_file_watcher.py -q` — 203 passed
- [ ] `python util/lint.py --all --fix` — all checks pass

Author: kovtcharov

hub/agents/python/emr/gaia_agent_emr/agent.py

@@ -264,7 +264,7 @@ def _print_file_listing(
                 size_str = "?"
 
             # Compute hash for status check
-            file_hash = compute_file_hash(f)
+            file_hash = compute_file_hash(f, allowed_dir=str(self._watch_dir))
             hash_display = file_hash[:8] + "..." if file_hash else "?"
 
             if file_hash and file_hash in processed_hashes:
@@ -336,7 +336,7 @@ def _process_existing_files(self) -> None:
         if new_count > 0:
             self.console.print_info(f"Processing {new_count} new file(s)...")
             for f in sorted(existing_files):
-                file_hash = compute_file_hash(f)
+                file_hash = compute_file_hash(f, allowed_dir=str(self._watch_dir))
                 if file_hash and file_hash not in processed_hashes:
                     self._on_file_created(f)
 
@@ -461,7 +461,7 @@ def _process_intake_form(self, file_path: str) -> Optional[Dict[str, Any]]:
 
             # Step 2: Check for duplicates
             self._emit_progress(filename, 2, total_steps, "Checking for duplicates")
-            file_hash = compute_file_hash(path)
+            file_hash = compute_file_hash(path, allowed_dir=str(self._watch_dir))
             if file_hash:
                 existing = self.query(
                     "SELECT id, first_name, last_name FROM patients WHERE file_hash = ?",

src/gaia/hub/installer.py

@@ -29,6 +29,7 @@
 import hashlib
 import json
 import os
+import re
 import shutil
 import stat
 import subprocess
@@ -122,17 +123,48 @@ class NotInstalledError(InstallError):
 # Paths
 # ---------------------------------------------------------------------------
 
+# Superset of the hub manifest id slug (manifest._ID_RE) that also tolerates
+# builtin/custom ids with uppercase, '.', or '_'. The security property is
+# that a valid id is exactly ONE path component: no '/', '\', ':', null, no
+# leading '.', so ``install_root / agent_id`` can never escape install_root.
+_AGENT_ID_SAFE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+
+
+def _require_safe_agent_id(agent_id: str) -> str:
+    """Return *agent_id* if it is safe to use as a directory name.
+
+    Agent ids reach this module from untrusted surfaces (the Agent UI's
+    ``POST /api/agents/install`` body, the downloaded hub catalog), and are
+    joined onto ``~/.gaia/agents/`` for destructive operations (rmtree,
+    move, write). Reject anything that is not a single sane path component.
+
+    Raises:
+        InstallError: If *agent_id* could traverse outside the install root.
+    """
+    if not isinstance(agent_id, str) or not _AGENT_ID_SAFE_RE.match(agent_id):
+        raise InstallError(
+            f"Invalid agent id {agent_id!r}: agent ids must start with a letter "
+            "or digit and contain only letters, digits, '.', '_' or '-' "
+            "(max 128 chars). Check the id against the hub catalog "
+            "('gaia agent list') or the package's gaia-agent.yaml."
+        )
+    return agent_id
+
 
 def default_install_root() -> Path:
     return Path.home() / ".gaia" / "agents"
 
 
 def agent_install_dir(agent_id: str, install_root: Optional[Path] = None) -> Path:
-    return (install_root or default_install_root()) / agent_id
+    return (install_root or default_install_root()) / _require_safe_agent_id(agent_id)
 
 
 def _backup_dir(agent_id: str, install_root: Optional[Path] = None) -> Path:
-    return (install_root or default_install_root()) / BACKUP_DIRNAME / agent_id
+    return (
+        (install_root or default_install_root())
+        / BACKUP_DIRNAME
+        / _require_safe_agent_id(agent_id)
+    )
 
 
 def _sentinel_path(agent_id: str, install_root: Optional[Path] = None) -> Path:
@@ -184,6 +216,23 @@ def read_sentinel(
     except (OSError, json.JSONDecodeError) as exc:
         logger.warning("installer: unreadable sentinel %s: %s", path, exc)
         return None
+    executable = data.get("executable", "")
+    # The sentinel is written from remote publish metadata; an executable
+    # with path separators could point health checks / the daemon outside
+    # the install dir. Treat it like any other corrupt sentinel.
+    if executable and (
+        not isinstance(executable, str)
+        or "/" in executable
+        or "\\" in executable
+        or Path(executable).name != executable
+    ):
+        logger.warning(
+            "installer: sentinel %s has unsafe executable %r (must be a bare "
+            "filename); treating agent as not installed — re-install it",
+            path,
+            executable,
+        )
+        return None
     return InstalledAgent(
         id=data.get("id", agent_id),
         version=data.get("version", ""),
@@ -207,6 +256,11 @@ def list_installed(install_root: Optional[Path] = None) -> Dict[str, InstalledAg
     for child in sorted(root.iterdir()):
         if not child.is_dir() or child.name == BACKUP_DIRNAME:
             continue
+        if not _AGENT_ID_SAFE_RE.match(child.name):
+            # Not a directory this installer could have created — skip it
+            # rather than crash the listing on stray junk in the root.
+            logger.warning("installer: ignoring non-agent directory %s", child)
+            continue
         if not (child / SENTINEL_NAME).exists():
             continue
         installed = read_sentinel(child.name, install_root)

src/gaia/hub/lifecycle.py

@@ -66,8 +66,16 @@ class LifecycleError(RuntimeError):
 
 
 def config_path(agent_id: str, install_root: Optional[Path] = None) -> Path:
-    """Path of the per-agent ``config.json`` (``~/.gaia/agents/<id>/config.json``)."""
-    return installer_mod.agent_install_dir(agent_id, install_root) / CONFIG_NAME
+    """Path of the per-agent ``config.json`` (``~/.gaia/agents/<id>/config.json``).
+
+    Raises:
+        LifecycleError: If *agent_id* is not a safe single path component
+            (path-traversal shaped ids are rejected by the installer).
+    """
+    try:
+        return installer_mod.agent_install_dir(agent_id, install_root) / CONFIG_NAME
+    except installer_mod.InstallError as exc:
+        raise LifecycleError(str(exc)) from exc
 
 
 # ---------------------------------------------------------------------------

src/gaia/ui/routers/hub.py

@@ -298,14 +298,20 @@ async def set_agent_config(agent_id: str, body: ConfigRequest):
 async def agent_health(agent_id: str, request: Request):
     """Health check: does the installed agent load + its entry point resolve?"""
     registry = _registry(request)
-    return lifecycle_mod.health_check(agent_id, registry=registry).to_dict()
+    try:
+        return lifecycle_mod.health_check(agent_id, registry=registry).to_dict()
+    except (installer_mod.InstallError, lifecycle_mod.LifecycleError) as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
 
 
 @router.get("/api/agents/{agent_id}/status")
 async def agent_status(agent_id: str, request: Request):
     """Aggregated status: installed version, health, config summary."""
     registry = _registry(request)
-    return lifecycle_mod.status(agent_id, registry=registry).to_dict()
+    try:
+        return lifecycle_mod.status(agent_id, registry=registry).to_dict()
+    except (installer_mod.InstallError, lifecycle_mod.LifecycleError) as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
 
 
 # ---------------------------------------------------------------------------

src/gaia/ui/utils.py

@@ -453,7 +453,8 @@ def safe_open_document(
 
     Validates that the path:
     - Is not a symlink (400 if symlink — checked before resolving)
-    - Is within the user home directory (403 if not)
+    - Is within the user home directory, both lexically and after resolving
+      every symlinked component (403 if not)
     - Has an allowed extension (400 if not)
     - Exists and is a regular file (404 / 400 if not)
 
@@ -476,7 +477,19 @@ def safe_open_document(
             detail=f"Access denied: path must be within home directory ({home})",
         )
 
-    # 2. Reject symlinks before resolving — lstat doesn't follow symlinks
+    # 2. Physical containment: the lexical check above can be defeated by an
+    # intermediate symlinked directory inside home that points outside
+    # (O_NOFOLLOW below only guards the FINAL path component). realpath
+    # resolves every component; require the result to stay under home.
+    real = os.path.realpath(str(raw))
+    home_prefix = str(home).rstrip(os.sep) + os.sep
+    if not real.startswith(home_prefix):
+        raise HTTPException(
+            status_code=403,
+            detail=f"Access denied: path must be within home directory ({home})",
+        )
+
+    # 3. Reject symlinks before resolving — lstat doesn't follow symlinks
     try:
         lst = os.lstat(str(raw))
         if stat.S_ISLNK(lst.st_mode):
@@ -489,17 +502,17 @@ def safe_open_document(
     except OSError as exc:
         raise HTTPException(status_code=400, detail=f"Cannot stat file: {exc}")
 
-    # 3. Now resolve (safe: we already confirmed it's not a symlink)
-    resolved = raw.resolve()
+    # 4. Open exactly the path we containment-checked (fully resolved)
+    resolved = Path(real)
 
-    # 4. Extension must be allowed
+    # 5. Extension must be allowed
     if resolved.suffix.lower() not in ALLOWED_EXTENSIONS:
         raise HTTPException(
             status_code=400,
             detail=f"File type not allowed: {resolved.suffix}. Allowed: {sorted(ALLOWED_EXTENSIONS)}",
         )
 
-    # 5. Open safely — use O_NOFOLLOW on POSIX to reject symlinks at kernel level
+    # 6. Open safely — use O_NOFOLLOW on POSIX to reject symlinks at kernel level
     flags = os.O_RDONLY
     if hasattr(os, "O_NOFOLLOW"):
         flags |= os.O_NOFOLLOW  # POSIX only; raises OSError(ELOOP) on symlinks
@@ -522,7 +535,7 @@ def safe_open_document(
 
     try:
         st = os.fstat(fd)
-        # 6. Must be a regular file (not dir, device, etc.)
+        # 7. Must be a regular file (not dir, device, etc.)
         if not stat.S_ISREG(st.st_mode):
             raise HTTPException(
                 status_code=400,

tests/unit/chat/ui/test_toctou.py

@@ -48,6 +48,33 @@ def test_safe_open_rejects_symlink(self, home_tmp_dir):
             or "symbolic" in exc_info.value.detail.lower()
         )
 
+    def test_safe_open_rejects_symlinked_directory_escape(self, home_tmp_dir):
+        """An intermediate symlinked dir inside home must not open files outside.
+
+        The lexical (abspath) home check passes for
+        ``~/.gaia_test_toctou/esc/secret.txt`` while the physical path lives
+        outside home — the realpath containment check must reject it with 403.
+        O_NOFOLLOW alone cannot catch this: it only guards the final component.
+        """
+        import tempfile
+
+        outside_dir = Path(tempfile.mkdtemp(prefix="gaia_toctou_outside_"))
+        try:
+            secret = outside_dir / "secret.txt"
+            secret.write_text("outside-home content")
+            escape_link = home_tmp_dir / "esc"
+            try:
+                escape_link.symlink_to(outside_dir, target_is_directory=True)
+            except OSError:
+                pytest.skip("Symlink creation requires elevated privileges on Windows")
+            attack_path = escape_link / "secret.txt"
+            with pytest.raises(HTTPException) as exc_info:
+                with safe_open_document(str(attack_path)):
+                    pass
+            assert exc_info.value.status_code == 403
+        finally:
+            shutil.rmtree(outside_dir, ignore_errors=True)
+
     def test_safe_open_rejects_missing_file(self, home_tmp_dir):
         """Non-existent file must return 404."""
         missing = home_tmp_dir / "nonexistent.txt"

tests/unit/test_file_watcher.py

@@ -86,6 +86,41 @@ def test_compute_file_hash_with_path_object(self):
         finally:
             temp_path.unlink()
 
+    def test_compute_file_hash_allowed_dir_rejects_traversal(self, tmp_path):
+        """A path outside allowed_dir must be rejected, including via '../'."""
+        allowed_dir = tmp_path / "watch"
+        allowed_dir.mkdir()
+        outside_dir = tmp_path / "outside"
+        outside_dir.mkdir()
+        secret = outside_dir / "secret.txt"
+        secret.write_bytes(b"do not hash me")
+
+        # Direct outside path
+        assert compute_file_hash(str(secret), allowed_dir=str(allowed_dir)) is None
+
+        # Traversal from within allowed_dir back out to the same file
+        traversal_path = allowed_dir / ".." / "outside" / "secret.txt"
+        assert (
+            compute_file_hash(str(traversal_path), allowed_dir=str(allowed_dir)) is None
+        )
+
+    def test_compute_file_hash_allowed_dir_rejects_absolute_escape(self, tmp_path):
+        """An absolute path outside allowed_dir is rejected, not silently hashed."""
+        allowed_dir = tmp_path / "watch"
+        allowed_dir.mkdir()
+        assert compute_file_hash("/etc/passwd", allowed_dir=str(allowed_dir)) is None
+
+    def test_compute_file_hash_allowed_dir_accepts_contained_path(self, tmp_path):
+        """A path genuinely inside allowed_dir still hashes normally."""
+        allowed_dir = tmp_path / "watch"
+        allowed_dir.mkdir()
+        f = allowed_dir / "form.pdf"
+        f.write_bytes(b"intake form bytes")
+
+        result = compute_file_hash(str(f), allowed_dir=str(allowed_dir))
+        assert result is not None
+        assert len(result) == 64
+
     def test_compute_bytes_hash_returns_sha256(self):
         """Test that compute_bytes_hash returns a valid SHA-256 hash."""
         result = compute_bytes_hash(b"test bytes")