diff --git a/AGENTS.md b/AGENTS.md index 1f15c38..84aa8ad 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Propagate unexpected errors to the CLI boundary. - Use documented defaults only for genuinely optional configuration. - Avoid compatibility shims without a named public or user configuration contract. +- Resolve every agent runtime directory as one physical child of `~/.elydora`; reject separators, traversal segments, cross-platform reserved names, and symbolic-link directories before writes or recursive removal. - Interpret `max_retries` as retries after the initial attempt and reject negative or non-integer values. - Automatically retry RFC-idempotent methods; retry non-idempotent methods only when the transport proves the request was never sent. - Honor valid `Retry-After` delay-seconds and HTTP-date values, and release retryable responses before sleeping. diff --git a/README.md b/README.md index 9f0414c..d6e0d5d 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,8 @@ async def main(): The SDK includes a CLI for installing audit hooks into AI coding agents. +Agent IDs map to one physical directory directly under `~/.elydora`; path separators, traversal segments, Windows reserved names, and symbolic-link directories are rejected. + ```bash elydora install \ --agent claudecode \ diff --git a/elydora/_runtime_paths.py b/elydora/_runtime_paths.py new file mode 100644 index 0000000..3fe0bfa --- /dev/null +++ b/elydora/_runtime_paths.py @@ -0,0 +1,68 @@ +"""Validated local runtime paths for agent installations.""" + +from __future__ import annotations + +import os +import re +import stat + + +_INVALID_FILENAME_CHARACTERS = re.compile(r'[<>:"|?*\x00-\x1f]') +_WINDOWS_DEVICE_NAME = re.compile( + r"^(?:con|prn|aux|nul|com[1-9¹²³]|lpt[1-9¹²³])(?:\.|$)", + re.IGNORECASE, +) + + +def resolve_agent_directory(root: str, agent_id: str) -> str: + """Resolve one cross-platform-safe agent directory directly under *root*.""" + if ( + not agent_id + or _INVALID_FILENAME_CHARACTERS.search(agent_id) + or "/" in agent_id + or "\\" in agent_id + or agent_id in {".", ".."} + or agent_id.endswith((".", " ")) + or _WINDOWS_DEVICE_NAME.match(agent_id) + ): + raise ValueError(f"Invalid agent ID for local storage: {agent_id!r}") + + resolved_root = os.path.abspath(root) + candidate = os.path.abspath(os.path.join(resolved_root, agent_id)) + try: + relative = os.path.relpath(candidate, resolved_root) + except ValueError as error: + raise ValueError( + f"Agent ID escapes the local storage directory: {agent_id!r}" + ) from error + if ( + relative in {"", ".", ".."} + or os.path.isabs(relative) + or relative.startswith(f"..{os.sep}") + or os.sep in relative + ): + raise ValueError( + f"Agent ID escapes the local storage directory: {agent_id!r}" + ) + return candidate + + +def ensure_private_directory(path: str) -> None: + """Create an owner-only directory and reject symbolic-link targets.""" + os.makedirs(path, mode=0o700, exist_ok=True) + metadata = os.lstat(path) + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"Private directory path is not a physical directory: {path}") + if os.name != "nt": + os.chmod(path, 0o700) + + +def require_physical_directory(path: str) -> bool: + """Return whether a path is an existing physical directory.""" + try: + metadata = os.lstat(path) + except FileNotFoundError: + return False + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"Agent runtime path is not a physical directory: {path}") + return True diff --git a/elydora/cli.py b/elydora/cli.py index ce0d8f3..7c22cb2 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -12,9 +12,14 @@ import json import os import shutil import sys -from typing import Dict, Type +from typing import Dict, NoReturn, Type from ._version import __version__ +from ._runtime_paths import ( + ensure_private_directory, + require_physical_directory, + resolve_agent_directory, +) from .crypto import get_public_key_base64url from .plugins.base import AgentPlugin, InstallConfig from .plugins.registry import SUPPORTED_AGENTS, get_agent_names @@ -55,6 +60,33 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { } +def _runtime_root() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def _exit_with_error(message: str) -> NoReturn: + print(f"Error: {message}", file=sys.stderr) + raise SystemExit(1) + + +def _resolve_agent_directory_or_exit(agent_id: str) -> str: + try: + return resolve_agent_directory(_runtime_root(), agent_id) + except ValueError as error: + _exit_with_error(str(error)) + + +def _read_runtime_config_or_exit(config_path: str) -> dict: + try: + with open(config_path, "r", encoding="utf-8") as file: + config = json.load(file) + except (OSError, json.JSONDecodeError) as error: + _exit_with_error(f"Read agent config at {config_path}: {error}") + if not isinstance(config, dict): + _exit_with_error(f"Agent config at {config_path} must contain a JSON object") + return config + + def _get_plugin(agent_name: str) -> AgentPlugin: """Instantiate the plugin for the given agent name.""" cls = PLUGIN_MAP.get(agent_name) @@ -68,6 +100,7 @@ def _get_plugin(agent_name: str) -> AgentPlugin: def cmd_install(args: argparse.Namespace) -> None: """Handle the 'install' subcommand.""" agent_name: str = args.agent + agent_dir = _resolve_agent_directory_or_exit(args.agent_id) # Derive public key to verify the private key is valid try: @@ -79,8 +112,8 @@ def cmd_install(args: argparse.Namespace) -> None: print(f"Verified key pair (public key: {pub[:16]}...)") # Create per-agent directory under ~/.elydora/{agent_id}/ - agent_dir = os.path.join(os.path.expanduser("~"), ".elydora", args.agent_id) - os.makedirs(agent_dir, exist_ok=True) + ensure_private_directory(_runtime_root()) + ensure_private_directory(agent_dir) # Generate and write the guard script guard_script_path = os.path.join(agent_dir, "guard.py") @@ -112,30 +145,48 @@ def cmd_install(args: argparse.Namespace) -> None: def cmd_uninstall(args: argparse.Namespace) -> None: """Handle the 'uninstall' subcommand.""" - plugin = _get_plugin(args.agent) - agent_id = getattr(args, "agent_id", None) or "" - plugin.uninstall(agent_id=agent_id) + explicit_agent_id = getattr(args, "agent_id", None) + if explicit_agent_id: + _resolve_agent_directory_or_exit(explicit_agent_id) # Determine which agent directory to remove - elydora_dir = os.path.join(os.path.expanduser("~"), ".elydora") - agent_id = getattr(args, "agent_id", None) + elydora_dir = _runtime_root() + require_physical_directory(elydora_dir) + agent_id = explicit_agent_id if not agent_id: # Scan ~/.elydora/*/config.json for matching agent_name pattern = os.path.join(elydora_dir, "*", "config.json") for config_path in glob.glob(pattern): - try: - with open(config_path, "r", encoding="utf-8") as f: - cfg = json.load(f) - if cfg.get("agent_name") == args.agent: - agent_id = cfg.get("agent_id") - break - except Exception: + cfg = _read_runtime_config_or_exit(config_path) + if cfg.get("agent_name") != args.agent: continue + stored_agent_id = cfg.get("agent_id") + if not isinstance(stored_agent_id, str): + _exit_with_error( + f"Agent config at {config_path} has an invalid agent_id" + ) + stored_directory = _resolve_agent_directory_or_exit(stored_agent_id) + if os.path.normcase(stored_directory) != os.path.normcase( + os.path.dirname(config_path) + ): + _exit_with_error( + f"Agent config at {config_path} crosses its runtime directory" + ) + agent_id = stored_agent_id + break + agent_dir = None + agent_directory_exists = False if agent_id: - agent_dir = os.path.join(elydora_dir, agent_id) - if os.path.isdir(agent_dir): + agent_dir = _resolve_agent_directory_or_exit(agent_id) + agent_directory_exists = require_physical_directory(agent_dir) + + plugin = _get_plugin(args.agent) + plugin.uninstall(agent_id=agent_id or "") + + if agent_dir: + if agent_directory_exists: shutil.rmtree(agent_dir) print(f" Removed agent directory: {agent_dir}") else: diff --git a/tests/test_runtime_paths.py b/tests/test_runtime_paths.py new file mode 100644 index 0000000..449523c --- /dev/null +++ b/tests/test_runtime_paths.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from elydora._runtime_paths import ( + ensure_private_directory, + require_physical_directory, + resolve_agent_directory, +) +from elydora import cli + + +@pytest.mark.parametrize( + "agent_id", + [ + "../escape", + "..\\escape", + "C:escape", + "agent.", + "agent ", + "CON", + "COM¹.log", + ".", + "..", + ], +) +def test_agent_directory_rejects_unsafe_cross_platform_names( + tmp_path: Path, + agent_id: str, +) -> None: + with pytest.raises(ValueError, match="agent ID"): + resolve_agent_directory(str(tmp_path / ".elydora"), agent_id) + + +def test_agent_directory_is_one_private_physical_child(tmp_path: Path) -> None: + root = tmp_path / ".elydora" + agent_dir = Path(resolve_agent_directory(str(root), "agent-1")) + + ensure_private_directory(str(root)) + ensure_private_directory(str(agent_dir)) + + assert agent_dir == root / "agent-1" + assert require_physical_directory(str(agent_dir)) is True + if os.name != "nt": + assert root.stat().st_mode & 0o777 == 0o700 + assert agent_dir.stat().st_mode & 0o777 == 0o700 + + +def test_private_directory_rejects_symbolic_links(tmp_path: Path) -> None: + target = tmp_path / "target" + target.mkdir() + link = tmp_path / "agent-link" + try: + link.symlink_to(target, target_is_directory=True) + except OSError as error: + pytest.skip(f"Directory symbolic links are unavailable: {error}") + + with pytest.raises(OSError, match="physical directory"): + ensure_private_directory(str(link)) + with pytest.raises(OSError, match="physical directory"): + require_physical_directory(str(link)) + + +def test_install_rejects_unsafe_agent_id_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + args = cli.build_parser().parse_args( + [ + "install", + "--agent", + "opencode", + "--org_id", + "org-1", + "--agent_id", + "../escape", + "--private_key", + "unused", + "--kid", + "key-1", + ] + ) + + with pytest.raises(SystemExit) as exc_info: + cli.cmd_install(args) + + assert exc_info.value.code == 1 + assert "Invalid agent ID for local storage" in capsys.readouterr().err + assert (tmp_path / ".elydora").exists() is False + + +def test_uninstall_validates_config_directory_before_plugin_changes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + runtime_dir = tmp_path / ".elydora" / "stored-directory" + runtime_dir.mkdir(parents=True) + (runtime_dir / "config.json").write_text( + json.dumps({"agent_name": "opencode", "agent_id": "different-agent"}), + encoding="utf-8", + ) + plugin_calls: list[str] = [] + + class Plugin: + def uninstall(self, agent_id: str = "") -> None: + plugin_calls.append(agent_id) + + monkeypatch.setattr(cli, "_get_plugin", lambda _name: Plugin()) + args = cli.build_parser().parse_args(["uninstall", "--agent", "opencode"]) + + with pytest.raises(SystemExit) as exc_info: + cli.cmd_uninstall(args) + + assert exc_info.value.code == 1 + assert "crosses its runtime directory" in capsys.readouterr().err + assert plugin_calls == [] + assert runtime_dir.is_dir()