fix(python-cli): confine agent runtime paths

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 16:49:26 -04:00
1 parent 9ae1b77e45
commit 1d83323d16
4 files changed
+436 -40

No files matched your search

+2
View File
@@ -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 \
+80
View File
@@ -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
+146 -40
View File
@@ -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
+208
View File
@@ -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()