diff --git a/src/openenv/auto/_discovery.py b/src/openenv/auto/_discovery.py index a41c0f802..abfc83a91 100644 --- a/src/openenv/auto/_discovery.py +++ b/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}") diff --git a/tests/envs/test_discovery.py b/tests/envs/test_discovery.py index d2ff24892..89baef8b2 100644 --- a/tests/envs/test_discovery.py +++ b/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."""