diff --git a/sdks/python/README.md b/sdks/python/README.md index ed42691..58367de 100644 --- a/sdks/python/README.md +++ b/sdks/python/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`; 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 \ --agent claudecode \ diff --git a/sdks/python/elydora/_runtime_paths.py b/sdks/python/elydora/_runtime_paths.py new file mode 100644 index 0000000..1dc25e7 --- /dev/null +++ b/sdks/python/elydora/_runtime_paths.py @@ -0,0 +1,80 @@ +"""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.startswith(" ") + 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 + + +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/sdks/python/elydora/cli.py b/sdks/python/elydora/cli.py index f7befaa..1c8545a 100644 --- a/sdks/python/elydora/cli.py +++ b/sdks/python/elydora/cli.py @@ -7,14 +7,20 @@ 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, 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 from .plugins.base import AgentPlugin, InstallConfig from .plugins.registry import SUPPORTED_AGENTS, get_agent_names @@ -55,6 +61,99 @@ 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)) + + +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) + 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 _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) @@ -68,6 +167,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,19 +179,16 @@ 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") 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 = { @@ -112,40 +209,49 @@ 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) + elydora_dir = _runtime_root() + agent_id = explicit_agent_id + if agent_id: + agent_dir = _resolve_agent_directory_or_exit(agent_id) + 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) - agent_id = getattr(args, "agent_id", None) or "" plugin.uninstall(agent_id=agent_id) - # Determine which agent directory to remove - elydora_dir = os.path.join(os.path.expanduser("~"), ".elydora") - agent_id = getattr(args, "agent_id", None) - - 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: - continue - - if agent_id: - agent_dir = os.path.join(elydora_dir, agent_id) - if os.path.isdir(agent_dir): - 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: @@ -187,7 +293,7 @@ def build_parser() -> argparse.ArgumentParser: install_parser.add_argument("--agent_id", required=True, help="Agent ID") install_parser.add_argument("--private_key", required=True, help="Base64url-encoded Ed25519 private key seed") install_parser.add_argument("--kid", required=True, help="Key ID") - install_parser.add_argument("--token", default="", help="API bearer token (optional)") + install_parser.add_argument("--token", default="", help="API token (from Console)") install_parser.add_argument("--base_url", default="https://api.elydora.com", help="API base URL") # uninstall diff --git a/sdks/python/tests/test_runtime_paths.py b/sdks/python/tests/test_runtime_paths.py new file mode 100644 index 0000000..1e25d2d --- /dev/null +++ b/sdks/python/tests/test_runtime_paths.py @@ -0,0 +1,208 @@ +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, + require_physical_file, + resolve_agent_directory, +) +from elydora import cli + + +@pytest.mark.parametrize( + "agent_id", + [ + "../escape", + "..\\escape", + "C:escape", + "agent.", + "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_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, + 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() + + +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()