fix(cli): protect Python install credentials

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 17:06:50 -04:00
1 parent 0c823e73e7
commit 6aba03b4c7
22 files changed
+787 -175

No files matched your search

+3
View File
@@ -42,6 +42,9 @@ This repository owns the published `elydora` package, its CLI, synchronous and a
- Use documented defaults only for genuinely optional configuration. - Use documented defaults only for genuinely optional configuration.
- Avoid compatibility shims without a named public or user configuration contract. - 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, 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. - 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.
- Accept CLI install credentials through hidden terminal input or physical owner-only credential files. Keep credentials out of process arguments, reject legacy secret options with redacted errors, and require one UTF-8 line of at most 64 KiB.
- Persist the signing key once at `~/.elydora/<agent-id>/private.key` with mode `0600`. Generated audit runtimes read and validate that physical file at execution time.
- Write runtime secrets, generated scripts, and provider configuration through same-directory temporary files followed by atomic replacement. Surface permission and replacement failures.
- Interpret `max_retries` as retries after the initial attempt and reject negative or non-integer values. - 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. - 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. - Honor valid `Retry-After` delay-seconds and HTTP-date values, and release retryable responses before sleeping.
+4 -1
View File
@@ -71,10 +71,13 @@ elydora install \
--agent claudecode \ --agent claudecode \
--org_id org-123 \ --org_id org-123 \
--agent_id agent-456 \ --agent_id agent-456 \
--private_key <key> \ --private_key_file /secure/path/private.key \
--token_file /secure/path/api.token \
--kid agent-456-key-v1 --kid agent-456-key-v1
``` ```
Credential options may be omitted in an interactive terminal; the CLI then reads the private key and optional API token through hidden prompts. Credential files must contain one UTF-8 line of at most 64 KiB. Unix credential files require owner-only permissions such as `chmod 600`. The installed signing key is stored once at `~/.elydora/<agent-id>/private.key` with mode `0600`, and generated audit hooks read that file at runtime.
Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions.
Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control. Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control.
+151
View File
@@ -0,0 +1,151 @@
"""Secure credential input for the Elydora CLI."""
from __future__ import annotations
import getpass
import os
import stat
import sys
import warnings
from typing import NamedTuple, Optional, Protocol
MAX_SECRET_FILE_BYTES = 64 * 1024
class InstallSecrets(NamedTuple):
private_key: str
token: Optional[str]
class SecretTerminal(Protocol):
@property
def interactive(self) -> bool:
...
def read_hidden(self, prompt: str) -> str:
...
class _DefaultTerminal:
@property
def interactive(self) -> bool:
return sys.stdin.isatty() and sys.stderr.isatty()
def read_hidden(self, prompt: str) -> str:
try:
with warnings.catch_warnings():
warnings.simplefilter("error", getpass.GetPassWarning)
return getpass.getpass(prompt=prompt, stream=sys.stderr)
except getpass.GetPassWarning as error:
raise RuntimeError(
"hidden input requires an interactive terminal"
) from error
except (EOFError, KeyboardInterrupt) as error:
raise RuntimeError("secret input cancelled") from error
def _parse_single_line_secret(raw: str, label: str, allow_empty: bool) -> str:
if len(raw.encode("utf-8")) > MAX_SECRET_FILE_BYTES:
raise ValueError(f"{label} exceeds {MAX_SECRET_FILE_BYTES} bytes")
if raw.endswith("\r\n"):
value = raw[:-2]
elif raw.endswith("\n"):
value = raw[:-1]
else:
value = raw
if not allow_empty and not value:
raise ValueError(f"{label} is empty")
if "\r" in value or "\n" in value or "\0" in value:
raise ValueError(f"{label} must contain exactly one line")
return value
def _validate_secret_file(
metadata: os.stat_result,
path: str,
label: str,
) -> None:
if not stat.S_ISREG(metadata.st_mode):
raise ValueError(f"{label} file is not a regular file: {path}")
if metadata.st_size > MAX_SECRET_FILE_BYTES:
raise ValueError(
f"{label} file exceeds {MAX_SECRET_FILE_BYTES} bytes: {path}"
)
if os.name != "nt" and metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
raise ValueError(
f"{label} file must be accessible only by its owner: {path}"
)
def _read_secret_file(path: str, label: str) -> str:
before = os.lstat(path)
_validate_secret_file(before, path, label)
flags = os.O_RDONLY
flags |= getattr(os, "O_BINARY", 0)
flags |= getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
try:
file = os.fdopen(descriptor, "rb")
descriptor = -1
with file:
after = os.fstat(file.fileno())
_validate_secret_file(after, path, label)
if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
raise OSError(f"{label} file changed while opening: {path}")
encoded = file.read(MAX_SECRET_FILE_BYTES + 1)
finally:
if descriptor >= 0:
os.close(descriptor)
if len(encoded) > MAX_SECRET_FILE_BYTES:
raise ValueError(
f"{label} file exceeds {MAX_SECRET_FILE_BYTES} bytes: {path}"
)
try:
raw = encoded.decode("utf-8")
except UnicodeDecodeError as error:
raise ValueError(f"{label} file must contain UTF-8 text: {path}") from error
return _parse_single_line_secret(raw, label, False)
def resolve_install_secrets(
*,
private_key_file: Optional[str],
token_file: Optional[str],
terminal: Optional[SecretTerminal] = None,
) -> InstallSecrets:
active_terminal = terminal if terminal is not None else _DefaultTerminal()
if private_key_file:
private_key = _read_secret_file(private_key_file, "private key")
elif active_terminal.interactive:
private_key = _parse_single_line_secret(
active_terminal.read_hidden("Private key: "),
"private key",
False,
)
else:
raise RuntimeError(
"private key input requires an interactive terminal or "
"--private_key_file <path>"
)
token: Optional[str]
if token_file:
token = _read_secret_file(token_file, "API token")
elif active_terminal.interactive:
token = (
_parse_single_line_secret(
active_terminal.read_hidden("API token (optional): "),
"API token",
True,
)
or None
)
else:
token = None
return InstallSecrets(private_key=private_key, token=token)
+48 -12
View File
@@ -10,10 +10,10 @@ import argparse
import json import json
import os import os
import shutil import shutil
import stat
import sys import sys
from typing import Dict, NamedTuple, NoReturn, Optional, Type from typing import Dict, NamedTuple, NoReturn, Optional, Type
from ._cli_secrets import resolve_install_secrets
from ._version import __version__ from ._version import __version__
from ._runtime_paths import ( from ._runtime_paths import (
ensure_private_directory, ensure_private_directory,
@@ -40,6 +40,7 @@ from .plugins.kimi import KimiPlugin
from .plugins.letta import LettaPlugin from .plugins.letta import LettaPlugin
from .plugins.opencode import OpenCodePlugin from .plugins.opencode import OpenCodePlugin
from .plugins.qwen import QwenPlugin from .plugins.qwen import QwenPlugin
from .plugins._file_io import write_text_atomic
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
@@ -60,6 +61,11 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
"qwen": QwenPlugin, "qwen": QwenPlugin,
} }
LEGACY_SECRET_OPTIONS = {
"--private_key": "--private_key_file",
"--token": "--token_file",
}
def _runtime_root() -> str: def _runtime_root() -> str:
return os.path.join(os.path.expanduser("~"), ".elydora") return os.path.join(os.path.expanduser("~"), ".elydora")
@@ -70,6 +76,17 @@ def _exit_with_error(message: str) -> NoReturn:
raise SystemExit(1) raise SystemExit(1)
def _reject_legacy_secret_arguments(arguments: list[str]) -> None:
for argument in arguments:
option = argument.split("=", 1)[0]
replacement = LEGACY_SECRET_OPTIONS.get(option)
if replacement:
_exit_with_error(
f"{option} exposes credentials in process arguments; "
f"use {replacement} or hidden terminal input"
)
def _resolve_agent_directory_or_exit(agent_id: str) -> str: def _resolve_agent_directory_or_exit(agent_id: str) -> str:
try: try:
return resolve_agent_directory(_runtime_root(), agent_id) return resolve_agent_directory(_runtime_root(), agent_id)
@@ -168,10 +185,19 @@ def cmd_install(args: argparse.Namespace) -> None:
"""Handle the 'install' subcommand.""" """Handle the 'install' subcommand."""
agent_name: str = args.agent agent_name: str = args.agent
agent_dir = _resolve_agent_directory_or_exit(args.agent_id) agent_dir = _resolve_agent_directory_or_exit(args.agent_id)
plugin = _get_plugin(agent_name)
try:
secrets = resolve_install_secrets(
private_key_file=args.private_key_file,
token_file=args.token_file,
)
except (OSError, RuntimeError, ValueError) as error:
_exit_with_error(str(error))
# Derive public key to verify the private key is valid # Derive public key to verify the private key is valid
try: try:
pub = get_public_key_base64url(args.private_key) pub = get_public_key_base64url(secrets.private_key)
except Exception as exc: except Exception as exc:
print(f"Error: Invalid private key — {exc}", file=sys.stderr) print(f"Error: Invalid private key — {exc}", file=sys.stderr)
sys.exit(1) sys.exit(1)
@@ -185,25 +211,26 @@ def cmd_install(args: argparse.Namespace) -> None:
# Generate and write the guard script # Generate and write the guard script
guard_script_path = os.path.join(agent_dir, "guard.py") guard_script_path = os.path.join(agent_dir, "guard.py")
guard_script = generate_guard_script(agent_name, args.agent_id) guard_script = generate_guard_script(agent_name, args.agent_id)
with open(guard_script_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(guard_script) guard_script_path,
if os.name != "nt": guard_script,
os.chmod(guard_script_path, os.stat(guard_script_path).st_mode | stat.S_IEXEC) 0o700,
"Elydora guard runtime",
)
print(f" Guard script: {guard_script_path}") print(f" Guard script: {guard_script_path}")
config: InstallConfig = { config: InstallConfig = {
"org_id": args.org_id, "org_id": args.org_id,
"agent_id": args.agent_id, "agent_id": args.agent_id,
"agent_name": agent_name, "agent_name": agent_name,
"private_key": args.private_key, "private_key": secrets.private_key,
"kid": args.kid, "kid": args.kid,
"base_url": args.base_url, "base_url": args.base_url,
"guard_script_path": guard_script_path, "guard_script_path": guard_script_path,
} }
if args.token: if secrets.token:
config["token"] = args.token config["token"] = secrets.token
plugin = _get_plugin(agent_name)
plugin.install(config) plugin.install(config)
@@ -291,9 +318,17 @@ def build_parser() -> argparse.ArgumentParser:
install_parser.add_argument("--agent", required=True, help="Agent name (e.g. claudecode, cursor)") install_parser.add_argument("--agent", required=True, help="Agent name (e.g. claudecode, cursor)")
install_parser.add_argument("--org_id", required=True, help="Organization ID") install_parser.add_argument("--org_id", required=True, help="Organization ID")
install_parser.add_argument("--agent_id", required=True, help="Agent ID") 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(
"--private_key_file",
default=None,
help="Owner-only file containing the Ed25519 private key seed",
)
install_parser.add_argument("--kid", required=True, help="Key ID") install_parser.add_argument("--kid", required=True, help="Key ID")
install_parser.add_argument("--token", default="", help="API token (from Console)") install_parser.add_argument(
"--token_file",
default=None,
help="Owner-only file containing the optional API token",
)
install_parser.add_argument("--base_url", default="https://api.elydora.com", help="API base URL") install_parser.add_argument("--base_url", default="https://api.elydora.com", help="API base URL")
# uninstall # uninstall
@@ -312,6 +347,7 @@ def build_parser() -> argparse.ArgumentParser:
def main() -> None: def main() -> None:
"""CLI entry point.""" """CLI entry point."""
_reject_legacy_secret_arguments(sys.argv[1:])
parser = build_parser() parser = build_parser()
args = parser.parse_args() args = parser.parse_args()
-1
View File
@@ -211,7 +211,6 @@ class AugmentPlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+21 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -40,34 +40,36 @@ class ClaudeCodePlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
# Write the hook script # Write the hook script
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) hook_path = self._hook_path_for(agent_id)
with open(hook_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) hook_path,
try: script,
os.chmod(hook_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -105,7 +107,7 @@ class ClaudeCodePlugin(AgentPlugin):
}) })
_save_json(SETTINGS_PATH, settings) _save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Claude Code.") print("Elydora hook installed for Claude Code.")
print(f" Hook script: {hook_path}") print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}") print(f" Settings: {SETTINGS_PATH}")
@@ -205,7 +207,4 @@ def _load_json(path: str) -> dict:
def _save_json(path: str, data: dict) -> None: def _save_json(path: str, data: dict) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True) write_json_atomic(path, data, 0o600, "Claude Code settings")
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
-1
View File
@@ -65,7 +65,6 @@ class ClinePlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
-1
View File
@@ -338,7 +338,6 @@ class CodexPlugin(AgentPlugin):
hook_script = generate_hook_script( hook_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+21 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -43,33 +43,35 @@ class CopilotPlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) hook_path = self._hook_path_for(agent_id)
with open(hook_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) hook_path,
try: script,
os.chmod(hook_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -103,7 +105,7 @@ class CopilotPlugin(AgentPlugin):
}) })
_save_json(settings_path, settings) _save_json(settings_path, settings)
print(f"Elydora hook installed for Copilot CLI.") print("Elydora hook installed for Copilot CLI.")
print(f" Hook script: {hook_path}") print(f" Hook script: {hook_path}")
print(f" Settings: {settings_path}") print(f" Settings: {settings_path}")
@@ -194,7 +196,4 @@ def _load_json(path: str) -> dict:
def _save_json(path: str, data: dict) -> None: def _save_json(path: str, data: dict) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True) write_json_atomic(path, data, 0o600, "GitHub Copilot hooks config")
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
+21 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -40,33 +40,35 @@ class CursorPlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) hook_path = self._hook_path_for(agent_id)
with open(hook_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) hook_path,
try: script,
os.chmod(hook_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -92,7 +94,7 @@ class CursorPlugin(AgentPlugin):
}) })
_save_json(SETTINGS_PATH, settings) _save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Cursor.") print("Elydora hook installed for Cursor.")
print(f" Hook script: {hook_path}") print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}") print(f" Settings: {SETTINGS_PATH}")
@@ -176,7 +178,4 @@ def _load_json(path: str) -> dict:
def _save_json(path: str, data: dict) -> None: def _save_json(path: str, data: dict) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True) write_json_atomic(path, data, 0o600, "Cursor hooks config")
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
-1
View File
@@ -108,7 +108,6 @@ class DroidPlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+21 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -40,33 +40,35 @@ class GeminiPlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) hook_path = self._hook_path_for(agent_id)
with open(hook_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) hook_path,
try: script,
os.chmod(hook_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -102,7 +104,7 @@ class GeminiPlugin(AgentPlugin):
}) })
_save_json(SETTINGS_PATH, settings) _save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Gemini CLI.") print("Elydora hook installed for Gemini CLI.")
print(f" Hook script: {hook_path}") print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}") print(f" Settings: {SETTINGS_PATH}")
@@ -201,7 +203,4 @@ def _load_json(path: str) -> dict:
def _save_json(path: str, data: dict) -> None: def _save_json(path: str, data: dict) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True) write_json_atomic(path, data, 0o600, "Gemini settings")
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
+1 -2
View File
@@ -9,7 +9,7 @@ import shlex
# subprocess provides argument quoting only; this module starts no process. # subprocess provides argument quoting only; this module starts no process.
import subprocess # nosec B404 import subprocess # nosec B404
import sys import sys
from typing import Any, Dict, List, Optional, Set, Tuple from typing import Any, Dict, List, Optional, Tuple
from ._file_io import ( from ._file_io import (
read_json as _read_json, read_json as _read_json,
@@ -362,7 +362,6 @@ class GrokPlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+43 -3
View File
@@ -14,7 +14,6 @@ def generate_hook_script(
*, *,
org_id: str, org_id: str,
agent_id: str, agent_id: str,
private_key: str,
kid: str, kid: str,
base_url: str, base_url: str,
) -> str: ) -> str:
@@ -38,6 +37,7 @@ import hashlib
import json import json
import math import math
import os import os
import stat
import struct import struct
import sys import sys
import time import time
@@ -47,14 +47,15 @@ import time
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
ORG_ID = {org_id!r} ORG_ID = {org_id!r}
AGENT_ID = {agent_id!r} AGENT_ID = {agent_id!r}
PRIVATE_KEY = {private_key!r}
KID = {kid!r} KID = {kid!r}
BASE_URL = {base_url!r} BASE_URL = {base_url!r}
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
PRIVATE_KEY_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "private.key")
CHAIN_STATE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "chain-state.json") CHAIN_STATE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "chain-state.json")
ERROR_LOG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "error.log") ERROR_LOG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "error.log")
ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
MAX_PRIVATE_KEY_BYTES = 64 * 1024
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Base64url helpers # Base64url helpers
@@ -134,6 +135,45 @@ def sign_ed25519(private_key_b64url, data):
key = Ed25519PrivateKey.from_private_bytes(seed) key = Ed25519PrivateKey.from_private_bytes(seed)
return base64url_encode(key.sign(data)) return base64url_encode(key.sign(data))
def read_private_key():
before = os.lstat(PRIVATE_KEY_PATH)
if not stat.S_ISREG(before.st_mode):
raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH)
if before.st_size > MAX_PRIVATE_KEY_BYTES:
raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH)
if os.name != "nt" and before.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH)
flags = os.O_RDONLY
flags |= getattr(os, "O_BINARY", 0)
flags |= getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(PRIVATE_KEY_PATH, flags)
try:
file = os.fdopen(descriptor, "rb")
descriptor = -1
with file:
after = os.fstat(file.fileno())
if not stat.S_ISREG(after.st_mode):
raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH)
if after.st_size > MAX_PRIVATE_KEY_BYTES:
raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH)
if os.name != "nt" and after.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH)
if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
raise OSError("Private key file changed while opening: " + PRIVATE_KEY_PATH)
encoded = file.read(MAX_PRIVATE_KEY_BYTES + 1)
finally:
if descriptor >= 0:
os.close(descriptor)
if len(encoded) > MAX_PRIVATE_KEY_BYTES:
raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH)
value = encoded.decode("utf-8")
if not value or "\\r" in value or "\\n" in value or "\\0" in value:
raise ValueError("Private key file must contain exactly one line")
return value
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# UUIDv7 # UUIDv7
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -259,7 +299,7 @@ def main():
}} }}
canonical = jcs_canonicalize(eor_without_sig) canonical = jcs_canonicalize(eor_without_sig)
signature = sign_ed25519(PRIVATE_KEY, canonical.encode("utf-8")) signature = sign_ed25519(read_private_key(), canonical.encode("utf-8"))
eor = dict(eor_without_sig) eor = dict(eor_without_sig)
eor["chain_hash"] = chain_hash eor["chain_hash"] = chain_hash
-1
View File
@@ -413,7 +413,6 @@ class KimiPlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
-1
View File
@@ -105,7 +105,6 @@ class KiroCliPlugin(AgentPlugin):
hook_script = generate_hook_script( hook_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+26 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -44,34 +44,36 @@ class KiroIdePlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
# Write the Python hook script # Write the Python hook script
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
script_path = self._hook_path_for(agent_id) script_path = self._hook_path_for(agent_id)
with open(script_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) script_path,
try: script,
os.chmod(script_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -93,13 +95,15 @@ class KiroIdePlugin(AgentPlugin):
}, },
} }
os.makedirs(KIRO_HOOK_DIR, exist_ok=True)
hook_path = self._kiro_hook_path() hook_path = self._kiro_hook_path()
with open(hook_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(hook_config, f, indent=2) hook_path,
f.write("\n") hook_config,
0o600,
"Kiro IDE hook definition",
)
print(f"Elydora hook installed for Kiro IDE.") print("Elydora hook installed for Kiro IDE.")
print(f" Hook script: {script_path}") print(f" Hook script: {script_path}")
print(f" Hook definition: {hook_path}") print(f" Hook definition: {hook_path}")
+21 -22
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
import json import json
import os import os
import stat
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -40,33 +40,35 @@ class LettaPlugin(AgentPlugin):
"agent_name": agent_name, "agent_name": agent_name,
} }
config_path = os.path.join(agent_dir, "config.json") config_path = os.path.join(agent_dir, "config.json")
with open(config_path, "w", encoding="utf-8") as f: write_json_atomic(
json.dump(config_data, f, indent=2) config_path,
f.write("\n") config_data,
0o600,
"Elydora runtime config",
)
# Write private key # Write private key
private_key_path = os.path.join(agent_dir, "private.key") private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(config.get("private_key", "")) private_key_path,
try: config.get("private_key", ""),
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) 0o600,
except Exception: "Elydora private key",
pass # chmod may fail on Windows )
script = generate_hook_script( script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) hook_path = self._hook_path_for(agent_id)
with open(hook_path, "w", encoding="utf-8") as f: write_text_atomic(
f.write(script) hook_path,
try: script,
os.chmod(hook_path, stat.S_IRWXU) 0o700,
except Exception: "Elydora audit runtime",
pass # chmod may fail on Windows )
guard_script_path = config.get("guard_script_path", "") guard_script_path = config.get("guard_script_path", "")
python_exe = sys.executable python_exe = sys.executable
@@ -104,7 +106,7 @@ class LettaPlugin(AgentPlugin):
}) })
_save_json(SETTINGS_PATH, settings) _save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Letta Code.") print("Elydora hook installed for Letta Code.")
print(f" Hook script: {hook_path}") print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}") print(f" Settings: {SETTINGS_PATH}")
@@ -203,7 +205,4 @@ def _load_json(path: str) -> dict:
def _save_json(path: str, data: dict) -> None: def _save_json(path: str, data: dict) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True) write_json_atomic(path, data, 0o600, "Letta settings")
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
+14 -17
View File
@@ -7,6 +7,7 @@ import json
import os import os
import sys import sys
from ._file_io import write_json_atomic, write_text_atomic
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
@@ -44,35 +45,41 @@ class OpenCodePlugin(AgentPlugin):
"token": config.get("token", ""), "token": config.get("token", ""),
"agent_name": config.get("agent_name", "opencode"), "agent_name": config.get("agent_name", "opencode"),
} }
_write_text( write_json_atomic(
os.path.join(agent_dir, "config.json"), os.path.join(agent_dir, "config.json"),
json.dumps(config_data, indent=2) + "\n", config_data,
0o600, 0o600,
"Elydora runtime config",
) )
_write_text( write_text_atomic(
os.path.join(agent_dir, "private.key"), os.path.join(agent_dir, "private.key"),
config.get("private_key", ""), config.get("private_key", ""),
0o600, 0o600,
"Elydora private key",
) )
hook_path = self._hook_path(agent_id) hook_path = self._hook_path(agent_id)
hook_script = generate_hook_script( hook_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
_write_text(hook_path, hook_script, 0o700) write_text_atomic(
hook_path,
hook_script,
0o700,
"Elydora audit runtime",
)
os.makedirs(PLUGIN_DIR, exist_ok=True) write_text_atomic(
_write_text(
self._plugin_path(), self._plugin_path(),
_generate_js_plugin( _generate_js_plugin(
hook_script_path=hook_path, hook_script_path=hook_path,
guard_script_path=guard_script_path, guard_script_path=guard_script_path,
), ),
0o700, 0o700,
"OpenCode Elydora plugin",
) )
print("Elydora hook installed for OpenCode.") print("Elydora hook installed for OpenCode.")
print(f" Plugin: {self._plugin_path()}") print(f" Plugin: {self._plugin_path()}")
@@ -105,16 +112,6 @@ class OpenCodePlugin(AgentPlugin):
details = f"Plugin: {plugin_path}" if installed else "Not installed" details = f"Plugin: {plugin_path}" if installed else "Not installed"
return PluginStatus(installed=installed, agent="opencode", details=details) return PluginStatus(installed=installed, agent="opencode", details=details)
def _write_text(path: str, content: str, mode: int) -> None:
with open(path, "w", encoding="utf-8") as file:
file.write(content)
try:
os.chmod(path, mode)
except OSError:
pass
def _generate_js_plugin(*, hook_script_path: str, guard_script_path: str) -> str: def _generate_js_plugin(*, hook_script_path: str, guard_script_path: str) -> str:
return f'''// Elydora Audit Plugin for OpenCode return f'''// Elydora Audit Plugin for OpenCode
// Generated by elydora-sdk — DO NOT EDIT // Generated by elydora-sdk — DO NOT EDIT
-1
View File
@@ -87,7 +87,6 @@ class QwenPlugin(AgentPlugin):
audit_script = generate_hook_script( audit_script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
+391
View File
@@ -0,0 +1,391 @@
from __future__ import annotations
import base64
import os
from pathlib import Path
from typing import List
import pytest
from elydora import _cli_secrets, cli
from elydora._cli_secrets import (
MAX_SECRET_FILE_BYTES,
InstallSecrets,
resolve_install_secrets,
)
from elydora.plugins._file_io import write_text_atomic
from elydora.plugins import (
claudecode,
copilot,
cursor,
gemini,
kiroide,
letta,
opencode,
)
from elydora.plugins.hook_template import generate_hook_script
PRIVATE_KEY = base64.urlsafe_b64encode(bytes([7]) * 32).rstrip(b"=").decode()
API_TOKEN = "ely_test_token"
class FakeTerminal:
def __init__(self, *, interactive: bool, answers: List[str]) -> None:
self.interactive = interactive
self.answers = answers
self.prompts: List[str] = []
def read_hidden(self, prompt: str) -> str:
self.prompts.append(prompt)
if not self.answers:
raise AssertionError("unexpected hidden prompt")
return self.answers.pop(0)
def write_secret(path: Path, value: str) -> None:
path.write_bytes(value.encode("utf-8"))
if os.name != "nt":
path.chmod(0o600)
def test_install_secrets_resolve_from_hidden_prompts() -> None:
terminal = FakeTerminal(
interactive=True,
answers=[PRIVATE_KEY, API_TOKEN],
)
secrets = resolve_install_secrets(
private_key_file=None,
token_file=None,
terminal=terminal,
)
assert secrets == InstallSecrets(PRIVATE_KEY, API_TOKEN)
assert terminal.prompts == ["Private key: ", "API token (optional): "]
def test_noninteractive_install_requires_private_key_file() -> None:
terminal = FakeTerminal(interactive=False, answers=[])
with pytest.raises(RuntimeError, match="--private_key_file <path>"):
resolve_install_secrets(
private_key_file=None,
token_file=None,
terminal=terminal,
)
assert terminal.prompts == []
def test_hidden_prompt_rejects_getpass_echo_fallback(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def fallback_getpass(prompt: str, stream: object) -> str:
_cli_secrets.warnings.warn(
"terminal echo is active",
_cli_secrets.getpass.GetPassWarning,
)
return PRIVATE_KEY
monkeypatch.setattr(_cli_secrets.getpass, "getpass", fallback_getpass)
with pytest.raises(RuntimeError, match="interactive terminal"):
_cli_secrets._DefaultTerminal().read_hidden("Private key: ")
def test_secret_files_allow_one_trailing_line_ending(tmp_path: Path) -> None:
private_key_file = tmp_path / "private.key"
token_file = tmp_path / "token"
write_secret(private_key_file, PRIVATE_KEY + "\r\n")
write_secret(token_file, API_TOKEN + "\n")
secrets = resolve_install_secrets(
private_key_file=str(private_key_file),
token_file=str(token_file),
terminal=FakeTerminal(interactive=False, answers=[]),
)
assert secrets == InstallSecrets(PRIVATE_KEY, API_TOKEN)
@pytest.mark.parametrize(
"content, message",
[
(PRIVATE_KEY + "\nsecond-line\n", "exactly one line"),
("", "private key is empty"),
("key\0value", "exactly one line"),
],
)
def test_private_key_file_rejects_invalid_content(
tmp_path: Path,
content: str,
message: str,
) -> None:
private_key_file = tmp_path / "private.key"
write_secret(private_key_file, content)
with pytest.raises(ValueError, match=message):
resolve_install_secrets(
private_key_file=str(private_key_file),
token_file=None,
terminal=FakeTerminal(interactive=False, answers=[]),
)
def test_private_key_file_rejects_oversized_content(tmp_path: Path) -> None:
private_key_file = tmp_path / "private.key"
write_secret(private_key_file, "a" * (MAX_SECRET_FILE_BYTES + 1))
with pytest.raises(ValueError, match="exceeds"):
resolve_install_secrets(
private_key_file=str(private_key_file),
token_file=None,
terminal=FakeTerminal(interactive=False, answers=[]),
)
def test_private_key_file_rejects_symbolic_links(tmp_path: Path) -> None:
target = tmp_path / "target.key"
write_secret(target, PRIVATE_KEY)
link = tmp_path / "linked.key"
try:
link.symlink_to(target)
except OSError as error:
pytest.skip(f"File symbolic links are unavailable: {error}")
with pytest.raises(ValueError, match="regular file"):
resolve_install_secrets(
private_key_file=str(link),
token_file=None,
terminal=FakeTerminal(interactive=False, answers=[]),
)
@pytest.mark.skipif(os.name == "nt", reason="POSIX permission contract")
def test_private_key_file_requires_owner_only_permissions(tmp_path: Path) -> None:
private_key_file = tmp_path / "private.key"
write_secret(private_key_file, PRIVATE_KEY)
private_key_file.chmod(0o640)
with pytest.raises(ValueError, match="accessible only by its owner"):
resolve_install_secrets(
private_key_file=str(private_key_file),
token_file=None,
terminal=FakeTerminal(interactive=False, answers=[]),
)
def test_secret_file_identity_is_stable_while_opening(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
private_key_file = tmp_path / "private.key"
replacement = tmp_path / "replacement.key"
write_secret(private_key_file, PRIVATE_KEY)
write_secret(replacement, PRIVATE_KEY)
original_open = os.open
swapped = False
def replace_then_open(path: str, flags: int) -> int:
nonlocal swapped
if path == str(private_key_file) and not swapped:
os.replace(replacement, private_key_file)
swapped = True
return original_open(path, flags)
monkeypatch.setattr(os, "open", replace_then_open)
with pytest.raises(OSError, match="changed while opening"):
resolve_install_secrets(
private_key_file=str(private_key_file),
token_file=None,
terminal=FakeTerminal(interactive=False, answers=[]),
)
@pytest.mark.parametrize(
"option, replacement",
[
("--private_key", "--private_key_file"),
("--token", "--token_file"),
],
)
def test_cli_rejects_legacy_secret_arguments_without_echoing_values(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
option: str,
replacement: str,
) -> None:
secret = "must-not-appear"
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setattr(
"sys.argv",
["elydora", "install", option, secret],
)
with pytest.raises(SystemExit) as exc_info:
cli.main()
assert exc_info.value.code == 1
error = capsys.readouterr().err
assert replacement in error
assert secret not in error
assert not (tmp_path / ".elydora").exists()
def test_cli_reads_credentials_from_files_before_installing(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
private_key_file = tmp_path / "private.key"
token_file = tmp_path / "token"
write_secret(private_key_file, PRIVATE_KEY)
write_secret(token_file, API_TOKEN)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
installed: list[dict[str, str]] = []
class Plugin:
def install(self, config: dict[str, str]) -> None:
installed.append(config)
monkeypatch.setattr(cli, "_get_plugin", lambda _name: Plugin())
args = cli.build_parser().parse_args(
[
"install",
"--agent",
"opencode",
"--org_id",
"org-1",
"--agent_id",
"agent-1",
"--private_key_file",
str(private_key_file),
"--token_file",
str(token_file),
"--kid",
"key-1",
]
)
cli.cmd_install(args)
assert installed[0]["private_key"] == PRIVATE_KEY
assert installed[0]["token"] == API_TOKEN
guard_path = tmp_path / ".elydora" / "agent-1" / "guard.py"
assert guard_path.is_file()
if os.name != "nt":
assert guard_path.stat().st_mode & 0o777 == 0o700
def test_generated_hook_reads_private_key_from_private_file(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
key_path = tmp_path / ".elydora" / "agent-1" / "private.key"
key_path.parent.mkdir(parents=True)
write_secret(key_path, PRIVATE_KEY)
script = generate_hook_script(
org_id="org-1",
agent_id="agent-1",
kid="key-1",
base_url="https://api.elydora.com",
)
namespace = {"__name__": "test_hook"}
exec(compile(script, "hook.py", "exec"), namespace)
assert namespace["read_private_key"]() == PRIVATE_KEY
assert PRIVATE_KEY not in script
assert "PRIVATE_KEY_PATH" in script
def test_atomic_secret_write_preserves_existing_file_on_replace_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
destination = tmp_path / "private.key"
destination.write_text("original", encoding="utf-8")
def fail_replace(_source: str, _destination: str) -> None:
raise OSError("replace failed")
monkeypatch.setattr(os, "replace", fail_replace)
with pytest.raises(OSError, match="replace failed"):
write_text_atomic(
str(destination),
"replacement",
0o600,
"Elydora private key",
)
assert destination.read_text(encoding="utf-8") == "original"
assert list(tmp_path.glob(".private.key.*.tmp")) == []
@pytest.mark.parametrize(
"module, plugin_type, agent_name",
[
(claudecode, claudecode.ClaudeCodePlugin, "claudecode"),
(copilot, copilot.CopilotPlugin, "copilot"),
(cursor, cursor.CursorPlugin, "cursor"),
(gemini, gemini.GeminiPlugin, "gemini"),
(kiroide, kiroide.KiroIdePlugin, "kiroide"),
(letta, letta.LettaPlugin, "letta"),
(opencode, opencode.OpenCodePlugin, "opencode"),
],
)
def test_legacy_plugins_persist_one_owner_only_private_key(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
module: object,
plugin_type: type,
agent_name: str,
) -> None:
case_root = tmp_path / agent_name
runtime_root = case_root / ".elydora"
guard_path = runtime_root / "agent-1" / "guard.py"
guard_path.parent.mkdir(parents=True)
guard_path.write_text("pass\n", encoding="utf-8")
monkeypatch.setattr(module, "ELYDORA_DIR", str(runtime_root))
if hasattr(module, "SETTINGS_PATH"):
monkeypatch.setattr(
module,
"SETTINGS_PATH",
str(case_root / "provider" / "settings.json"),
)
elif module is kiroide:
monkeypatch.setattr(module, "KIRO_HOOK_DIR", str(case_root / "kiro-hooks"))
elif module is opencode:
monkeypatch.setattr(module, "PLUGIN_DIR", str(case_root / "plugins"))
elif module is copilot:
case_root.mkdir(parents=True, exist_ok=True)
monkeypatch.chdir(case_root)
plugin_type().install(
{
"org_id": "org-1",
"agent_id": "agent-1",
"agent_name": agent_name,
"private_key": PRIVATE_KEY,
"kid": "key-1",
"token": API_TOKEN,
"base_url": "https://api.elydora.com",
"guard_script_path": str(guard_path),
}
)
private_key_path = runtime_root / "agent-1" / "private.key"
hook_path = runtime_root / "agent-1" / "hook.py"
assert private_key_path.read_text(encoding="utf-8") == PRIVATE_KEY
assert PRIVATE_KEY not in hook_path.read_text(encoding="utf-8")
if os.name != "nt":
assert private_key_path.stat().st_mode & 0o777 == 0o600
assert hook_path.stat().st_mode & 0o777 == 0o700
+1 -1
View File
@@ -96,7 +96,7 @@ def test_install_rejects_unsafe_agent_id_before_writes(
"org-1", "org-1",
"--agent_id", "--agent_id",
"../escape", "../escape",
"--private_key", "--private_key_file",
"unused", "unused",
"--kid", "--kid",
"key-1", "key-1",