fix(discovery): store env discovery cache per-user, not in world-writable tmp

src/openenv/auto/_discovery.py

@@ -20,8 +20,9 @@
 import importlib.resources
 import json
 import logging
+import os
 import re
-import tempfile
+import stat
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Any, Type
@@ -336,6 +337,46 @@ def _create_env_info_from_package(
     )
 
 
+def _default_cache_file() -> Path:
+    """
+    Return the per-user discovery cache file path.
+
+    Uses a per-user cache directory (``$XDG_CACHE_HOME`` or ``~/.cache``) rather than a
+    shared, world-writable temporary directory. A fixed path under the shared temp dir
+    lets another local user pre-create the cache file and redirect discovery to
+    attacker-controlled import paths (`import_module` on a cached `client_module_path`).
+    """
+    base = os.environ.get("XDG_CACHE_HOME")
+    root = Path(base) if base else Path.home() / ".cache"
+    return root / "openenv" / "discovery_cache.json"
+
+
+def _is_trusted_cache_file(path: Path) -> bool:
+    """
+    Return whether *path* is safe to load.
+
+    On POSIX the file must be owned by the current user and not writable by group or
+    others, so a cache file planted by another user is ignored rather than trusted.
+
+    Args:
+        path (`Path`):
+            The cache file to check.
+
+    Returns:
+        `bool`: `True` if the file is owned by the current user and not
+        group/other-writable (always `True` on non-POSIX platforms).
+    """
+    if os.name != "posix":
+        return True
+    try:
+        info = path.stat()
+    except OSError:
+        return False
+    return info.st_uid == os.getuid() and not (
+        info.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
+    )
+
+
 class EnvironmentDiscovery:
     """
     Auto-discovery system for OpenEnv environments using installed packages.
@@ -346,7 +387,7 @@ class EnvironmentDiscovery:
     def __init__(self):
         """Initialize discovery system."""
         self._cache: dict[str, EnvironmentInfo] | None = None
-        self._cache_file = Path(tempfile.gettempdir()) / "openenv_discovery_cache.json"
+        self._cache_file = _default_cache_file()
 
     def _discover_installed_packages(self) -> dict[str, EnvironmentInfo]:
         """
@@ -411,6 +452,16 @@ def _load_cache(self) -> dict[str, EnvironmentInfo] | None:
         if not self._cache_file.exists():
             return None
 
+        # Only trust a cache file owned by the current user. This prevents another
+        # local user from planting a file that would redirect discovery (and the
+        # subsequent import_module) to attacker-controlled modules/classes.
+        if not _is_trusted_cache_file(self._cache_file):
+            logger.warning(
+                f"Ignoring discovery cache {self._cache_file}: not owned by the "
+                "current user or writable by group/others."
+            )
+            return None
+
         try:
             with open(self._cache_file, "r") as f:
                 cache_data = json.load(f)
@@ -437,8 +488,12 @@ def _save_cache(self, environments: dict[str, EnvironmentInfo]) -> None:
             for env_key, env_info in environments.items():
                 cache_data[env_key] = asdict(env_info)
 
+            self._cache_file.parent.mkdir(parents=True, exist_ok=True)
             with open(self._cache_file, "w") as f:
                 json.dump(cache_data, f, indent=2)
+            # Restrict to the owner so the cache cannot be tampered with by others.
+            if os.name == "posix":
+                os.chmod(self._cache_file, 0o600)
 
         except Exception as e:
             logger.warning(f"Failed to save discovery cache: {e}")

tests/envs/test_discovery.py

@@ -12,12 +12,18 @@
 5. Helper functions (_normalize_env_name, _is_hub_url, etc.)
 """
 
+import json
+import os
+import stat
+import tempfile
 from unittest.mock import Mock, patch
 
 from openenv.auto._discovery import (
     _create_env_info_from_package,
+    _default_cache_file,
     _infer_class_name,
     _is_hub_url,
+    _is_trusted_cache_file,
     _normalize_env_name,
     EnvironmentDiscovery,
     EnvironmentInfo,
@@ -304,6 +310,80 @@ def test_cache_management(self):
         assert not discovery._cache_file.exists()
 
 
+class TestCacheSecurity:
+    """The discovery cache must not be plantable/redirectable by another local user.
+
+    A world-writable shared path let a local attacker pre-create the cache and
+    redirect discovery's later ``import_module`` to attacker-chosen modules/classes
+    (CWE-377 insecure temp file + CWE-427 uncontrolled load path).
+    """
+
+    def test_cache_file_is_per_user_not_shared_tmp(self):
+        path = _default_cache_file()
+        assert tempfile.gettempdir() not in str(path)
+        assert path.parent.name == "openenv"
+
+    def test_world_writable_cache_is_not_trusted(self, tmp_path):
+        f = tmp_path / "cache.json"
+        f.write_text("{}")
+        os.chmod(f, 0o666)
+        # Rejected on POSIX; non-POSIX has no ownership model so it stays trusted.
+        assert _is_trusted_cache_file(f) is (os.name != "posix")
+
+    def test_owner_only_cache_is_trusted(self, tmp_path):
+        f = tmp_path / "cache.json"
+        f.write_text("{}")
+        os.chmod(f, 0o600)
+        assert _is_trusted_cache_file(f) is True
+
+    def test_load_cache_ignores_world_writable_file(self, tmp_path):
+        planted = tmp_path / "cache.json"
+        planted.write_text(
+            json.dumps(
+                {
+                    "evil": {
+                        "env_key": "evil",
+                        "name": "evil",
+                        "package_name": "openenv-evil",
+                        "version": "1.0.0",
+                        "description": "",
+                        "client_module_path": "os",
+                        "client_class_name": "system",
+                        "action_class_name": "A",
+                        "observation_class_name": "O",
+                        "default_image": "i",
+                    }
+                }
+            )
+        )
+        os.chmod(planted, 0o666)
+        discovery = EnvironmentDiscovery()
+        discovery._cache_file = planted
+        if os.name == "posix":
+            assert discovery._load_cache() is None
+
+    def test_saved_cache_is_owner_only(self, tmp_path):
+        discovery = EnvironmentDiscovery()
+        discovery._cache_file = tmp_path / "sub" / "cache.json"
+        env = EnvironmentInfo(
+            env_key="t",
+            name="t",
+            package_name="openenv-t",
+            version="1.0.0",
+            description="",
+            client_module_path="t.client",
+            client_class_name="T",
+            action_class_name="A",
+            observation_class_name="O",
+            default_image="i",
+        )
+        discovery._save_cache({"t": env})
+        assert discovery._cache_file.exists()
+        if os.name == "posix":
+            assert stat.S_IMODE(discovery._cache_file.stat().st_mode) == 0o600
+        assert discovery._load_cache() is not None
+
+
 class TestGlobalDiscovery:
     """Test global discovery instance management."""