diff --git a/AGENTS.md b/AGENTS.md index 84aa8ad..bb7088d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +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. +- Resolve every agent runtime directory as one physical child of `~/.elydora`; reject separators, traversal segments, cross-platform reserved names, symbolic-link directories, and linked identity configs before writes or recursive removal. Validate stored directory identity before changing host CLI configuration, and require an explicit agent ID when discovery is ambiguous. - 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 d6e0d5d..1dc8c87 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ 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. +Agent IDs map to one physical directory directly under `~/.elydora`; portable filename rules and physical directory/config checks apply before writes or recursive removal. Ambiguous uninstall discovery requires an explicit agent ID. ```bash elydora install \ diff --git a/elydora/_runtime_paths.py b/elydora/_runtime_paths.py index 03a2acf..1dc25e7 100644 --- a/elydora/_runtime_paths.py +++ b/elydora/_runtime_paths.py @@ -67,3 +67,14 @@ def require_physical_directory(path: str) -> bool: 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 + + +def require_physical_file(path: str) -> bool: + """Return whether a path is an existing physical regular file.""" + try: + metadata = os.lstat(path) + except FileNotFoundError: + return False + if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"Agent runtime config is not a physical file: {path}") + return True diff --git a/elydora/cli.py b/elydora/cli.py index 7c22cb2..1c8545a 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -7,17 +7,18 @@ Uses only stdlib argparse — zero external dependencies for the CLI itself. from __future__ import annotations import argparse -import glob import json import os import shutil +import stat import sys -from typing import Dict, NoReturn, Type +from typing import Dict, NamedTuple, NoReturn, Optional, Type from ._version import __version__ from ._runtime_paths import ( ensure_private_directory, require_physical_directory, + require_physical_file, resolve_agent_directory, ) from .crypto import get_public_key_base64url @@ -76,7 +77,27 @@ def _resolve_agent_directory_or_exit(agent_id: str) -> str: _exit_with_error(str(error)) -def _read_runtime_config_or_exit(config_path: str) -> dict: +class _InstalledAgent(NamedTuple): + agent_id: str + agent_name: str + config_path: str + + +def _require_physical_directory_or_exit(path: str) -> bool: + try: + return require_physical_directory(path) + except OSError as error: + _exit_with_error(str(error)) + + +def _require_physical_file_or_exit(path: str) -> bool: + try: + return require_physical_file(path) + except OSError as error: + _exit_with_error(str(error)) + + +def _read_runtime_config_or_exit(config_path: str) -> Dict[str, object]: try: with open(config_path, "r", encoding="utf-8") as file: config = json.load(file) @@ -87,6 +108,52 @@ def _read_runtime_config_or_exit(config_path: str) -> dict: return config +def _read_installed_agent_or_exit(agent_id: str) -> Optional[_InstalledAgent]: + agent_directory = _resolve_agent_directory_or_exit(agent_id) + if not _require_physical_directory_or_exit(agent_directory): + return None + + config_path = os.path.join(agent_directory, "config.json") + if not _require_physical_file_or_exit(config_path): + return None + config = _read_runtime_config_or_exit(config_path) + stored_agent_id = config.get("agent_id") + stored_agent_name = config.get("agent_name") + if not isinstance(stored_agent_id, str) or not isinstance( + stored_agent_name, str + ): + _exit_with_error(f"Agent config at {config_path} has an invalid identity") + if stored_agent_id != agent_id: + _exit_with_error( + f"Agent config at {config_path} crosses its runtime directory" + ) + _resolve_agent_directory_or_exit(stored_agent_id) + return _InstalledAgent(agent_id, stored_agent_name, config_path) + + +def _discover_installed_agents_or_exit() -> list[_InstalledAgent]: + runtime_root = _runtime_root() + if not _require_physical_directory_or_exit(runtime_root): + return [] + + installed_agents: list[_InstalledAgent] = [] + try: + with os.scandir(runtime_root) as entries: + for entry in entries: + if entry.is_symlink(): + _exit_with_error( + f"Agent runtime path is not a physical directory: {entry.path}" + ) + if not entry.is_dir(follow_symlinks=False): + continue + installed_agent = _read_installed_agent_or_exit(entry.name) + if installed_agent is not None: + installed_agents.append(installed_agent) + except OSError as error: + _exit_with_error(f"Scan agent runtime root at {runtime_root}: {error}") + return installed_agents + + def _get_plugin(agent_name: str) -> AgentPlugin: """Instantiate the plugin for the given agent name.""" cls = PLUGIN_MAP.get(agent_name) @@ -120,11 +187,8 @@ def cmd_install(args: argparse.Namespace) -> None: guard_script = generate_guard_script(agent_name, args.agent_id) with open(guard_script_path, "w", encoding="utf-8") as f: f.write(guard_script) - try: - import stat + if os.name != "nt": os.chmod(guard_script_path, os.stat(guard_script_path).st_mode | stat.S_IEXEC) - except Exception: - pass # chmod may fail on Windows print(f" Guard script: {guard_script_path}") config: InstallConfig = { @@ -146,57 +210,48 @@ def cmd_install(args: argparse.Namespace) -> None: def cmd_uninstall(args: argparse.Namespace) -> None: """Handle the 'uninstall' subcommand.""" 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 = _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): - 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 = _resolve_agent_directory_or_exit(agent_id) - agent_directory_exists = require_physical_directory(agent_dir) + runtime_root_exists = _require_physical_directory_or_exit(elydora_dir) + agent_directory_exists = ( + _require_physical_directory_or_exit(agent_dir) + if runtime_root_exists + else False + ) + if agent_directory_exists: + installed_agent = _read_installed_agent_or_exit(agent_id) + if installed_agent and installed_agent.agent_name != args.agent: + _exit_with_error( + f"Agent runtime {agent_id} belongs to " + f"{installed_agent.agent_name}, not {args.agent}" + ) + else: + matches = [ + installed_agent + for installed_agent in _discover_installed_agents_or_exit() + if installed_agent.agent_name == args.agent + ] + if not matches: + _exit_with_error( + f"No installed agent found for {args.agent!r}; pass --agent_id explicitly" + ) + if len(matches) > 1: + _exit_with_error( + f"Multiple installed agents found for {args.agent!r}; " + "pass --agent_id explicitly" + ) + agent_id = matches[0].agent_id + agent_dir = _resolve_agent_directory_or_exit(agent_id) + agent_directory_exists = True plugin = _get_plugin(args.agent) - plugin.uninstall(agent_id=agent_id or "") + plugin.uninstall(agent_id=agent_id) - if agent_dir: - if agent_directory_exists: - shutil.rmtree(agent_dir) - print(f" Removed agent directory: {agent_dir}") - else: - # Fallback: try to remove legacy guard script location - guard_script_path = os.path.join( - elydora_dir, "hooks", f"{args.agent}-guard.py" - ) - if os.path.exists(guard_script_path): - os.remove(guard_script_path) - print(f" Removed guard script: {guard_script_path}") + if agent_directory_exists: + shutil.rmtree(agent_dir) + print(f" Removed agent directory: {agent_dir}") def cmd_status(args: argparse.Namespace) -> None: diff --git a/tests/test_runtime_paths.py b/tests/test_runtime_paths.py index d70cafe..1e25d2d 100644 --- a/tests/test_runtime_paths.py +++ b/tests/test_runtime_paths.py @@ -9,6 +9,7 @@ import pytest from elydora._runtime_paths import ( ensure_private_directory, require_physical_directory, + require_physical_file, resolve_agent_directory, ) from elydora import cli @@ -66,6 +67,19 @@ def test_private_directory_rejects_symbolic_links(tmp_path: Path) -> None: require_physical_directory(str(link)) +def test_runtime_config_rejects_symbolic_links(tmp_path: Path) -> None: + target = tmp_path / "target.json" + target.write_text("{}", encoding="utf-8") + link = tmp_path / "config.json" + try: + link.symlink_to(target) + except OSError as error: + pytest.skip(f"File symbolic links are unavailable: {error}") + + with pytest.raises(OSError, match="physical file"): + require_physical_file(str(link)) + + def test_install_rejects_unsafe_agent_id_before_writes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -126,3 +140,69 @@ def test_uninstall_validates_config_directory_before_plugin_changes( assert "crosses its runtime directory" in capsys.readouterr().err assert plugin_calls == [] assert runtime_dir.is_dir() + + +def test_uninstall_requires_explicit_id_for_ambiguous_runtimes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + for agent_id in ("agent-1", "agent-2"): + runtime_dir = tmp_path / ".elydora" / agent_id + runtime_dir.mkdir(parents=True) + (runtime_dir / "config.json").write_text( + json.dumps({"agent_name": "opencode", "agent_id": agent_id}), + 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 "Multiple installed agents" in capsys.readouterr().err + assert plugin_calls == [] + assert (tmp_path / ".elydora" / "agent-1").is_dir() + assert (tmp_path / ".elydora" / "agent-2").is_dir() + + +def test_explicit_uninstall_validates_agent_ownership_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" / "agent-1" + runtime_dir.mkdir(parents=True) + (runtime_dir / "config.json").write_text( + json.dumps({"agent_name": "codex", "agent_id": "agent-1"}), + 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", "--agent_id", "agent-1"] + ) + + with pytest.raises(SystemExit) as exc_info: + cli.cmd_uninstall(args) + + assert exc_info.value.code == 1 + assert "belongs to codex" in capsys.readouterr().err + assert plugin_calls == [] + assert runtime_dir.is_dir()