mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-11 12:48:54 +08:00
Compare commits
1 Commits
bb/model-r
...
feat/plugi
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cc60cbfeb5 |
@@ -135,34 +135,89 @@ def _sanitize_plugin_name(
|
||||
return target
|
||||
|
||||
|
||||
def _resolve_git_url(identifier: str) -> str:
|
||||
"""Turn an identifier into a cloneable Git URL.
|
||||
def _resolve_git_url(identifier: str) -> tuple[str, Optional[str]]:
|
||||
"""Turn an identifier into a cloneable Git URL and optional subdirectory.
|
||||
|
||||
Returns ``(git_url, subdir)`` where ``subdir`` is the path within the
|
||||
cloned repository that contains the plugin (``None`` when the plugin lives
|
||||
at the repo root).
|
||||
|
||||
Accepted formats:
|
||||
- Full URL: https://github.com/owner/repo.git
|
||||
- Full URL: git@github.com:owner/repo.git
|
||||
- Full URL: ssh://git@github.com/owner/repo.git
|
||||
- Shorthand: owner/repo → https://github.com/owner/repo.git
|
||||
- Shorthand w/ subdir: owner/repo/path/to/plugin
|
||||
→ (https://github.com/owner/repo.git, "path/to/plugin")
|
||||
- Full URL w/ subdir (``.git`` boundary):
|
||||
https://github.com/owner/repo.git/path/to/plugin
|
||||
→ (https://github.com/owner/repo.git, "path/to/plugin")
|
||||
- Any URL w/ explicit subdir fragment (works for every scheme, incl.
|
||||
``file://`` and ssh): <url>#path/to/plugin
|
||||
→ (<url>, "path/to/plugin")
|
||||
|
||||
NOTE: ``http://`` and ``file://`` schemes are accepted but will trigger a
|
||||
security warning at install time.
|
||||
"""
|
||||
# Already a URL
|
||||
# Already a URL.
|
||||
if identifier.startswith(("https://", "http://", "git@", "ssh://", "file://")):
|
||||
return identifier
|
||||
# Explicit ``#subdir`` fragment — unambiguous for any scheme.
|
||||
if "#" in identifier:
|
||||
git_url, _, frag = identifier.partition("#")
|
||||
return git_url, (frag.strip("/") or None)
|
||||
# Natural ``.git/`` boundary (GitHub-style URLs).
|
||||
marker = ".git/"
|
||||
idx = identifier.find(marker)
|
||||
if idx != -1:
|
||||
git_url = identifier[: idx + len(".git")]
|
||||
subdir = identifier[idx + len(marker) :].strip("/")
|
||||
return git_url, (subdir or None)
|
||||
return identifier, None
|
||||
|
||||
# owner/repo shorthand
|
||||
parts = identifier.strip("/").split("/")
|
||||
if len(parts) == 2:
|
||||
owner, repo = parts
|
||||
return f"https://github.com/{owner}/{repo}.git"
|
||||
# owner/repo[/subdir...] shorthand
|
||||
parts = [p for p in identifier.strip("/").split("/") if p]
|
||||
if len(parts) >= 2:
|
||||
owner, repo = parts[0], parts[1]
|
||||
subdir = "/".join(parts[2:]).strip("/")
|
||||
git_url = f"https://github.com/{owner}/{repo}.git"
|
||||
return git_url, (subdir or None)
|
||||
|
||||
raise ValueError(
|
||||
f"Invalid plugin identifier: '{identifier}'. "
|
||||
"Use a Git URL or owner/repo shorthand."
|
||||
"Use a Git URL or 'owner/repo' shorthand (optionally with a subdirectory: "
|
||||
"'owner/repo/path/to/plugin')."
|
||||
)
|
||||
|
||||
|
||||
def _resolve_subdir_within(clone_root: Path, subdir: str) -> Path:
|
||||
"""Resolve ``subdir`` inside ``clone_root``, rejecting path traversal.
|
||||
|
||||
Guards against ``..`` segments, absolute paths, and symlinks that would
|
||||
escape the cloned repository. Returns the resolved directory path.
|
||||
Raises ``PluginOperationError`` if the path escapes the clone, doesn't
|
||||
exist, or is not a directory.
|
||||
"""
|
||||
clone_root = clone_root.resolve()
|
||||
candidate = (clone_root / subdir).resolve()
|
||||
|
||||
# The resolved candidate must stay within the clone root.
|
||||
if candidate != clone_root and clone_root not in candidate.parents:
|
||||
raise PluginOperationError(
|
||||
f"Plugin subdirectory '{subdir}' escapes the repository.",
|
||||
)
|
||||
|
||||
if not candidate.exists():
|
||||
raise PluginOperationError(
|
||||
f"Plugin subdirectory '{subdir}' does not exist in the repository.",
|
||||
)
|
||||
if not candidate.is_dir():
|
||||
raise PluginOperationError(
|
||||
f"Plugin subdirectory '{subdir}' is not a directory.",
|
||||
)
|
||||
|
||||
return candidate
|
||||
|
||||
|
||||
def _repo_name_from_url(url: str) -> str:
|
||||
"""Extract the repo name from a Git URL for the plugin directory name."""
|
||||
# Strip trailing .git and slashes
|
||||
@@ -372,14 +427,14 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
|
||||
import tempfile
|
||||
|
||||
try:
|
||||
git_url = _resolve_git_url(identifier)
|
||||
git_url, subdir = _resolve_git_url(identifier)
|
||||
except ValueError as e:
|
||||
raise PluginOperationError(str(e)) from e
|
||||
|
||||
plugins_dir = _plugins_dir()
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp_target = Path(tmp) / "plugin"
|
||||
tmp_clone = Path(tmp) / "plugin"
|
||||
|
||||
git_exe = _resolve_git_executable()
|
||||
if not git_exe:
|
||||
@@ -387,7 +442,7 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[git_exe, "clone", "--depth", "1", git_url, str(tmp_target)],
|
||||
[git_exe, "clone", "--depth", "1", git_url, str(tmp_clone)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
@@ -405,8 +460,16 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s
|
||||
err = (result.stderr or result.stdout or "").strip()
|
||||
raise PluginOperationError(f"Git clone failed:\n{err}")
|
||||
|
||||
# Resolve the directory within the clone that holds the plugin.
|
||||
if subdir:
|
||||
tmp_target = _resolve_subdir_within(tmp_clone, subdir)
|
||||
else:
|
||||
tmp_target = tmp_clone
|
||||
|
||||
manifest = _read_manifest(tmp_target)
|
||||
plugin_name = manifest.get("name") or _repo_name_from_url(git_url)
|
||||
plugin_name = manifest.get("name") or (
|
||||
subdir.rstrip("/").rsplit("/", 1)[-1] if subdir else _repo_name_from_url(git_url)
|
||||
)
|
||||
|
||||
try:
|
||||
target = _sanitize_plugin_name(plugin_name, plugins_dir)
|
||||
@@ -471,7 +534,7 @@ def cmd_install(
|
||||
console = Console()
|
||||
|
||||
try:
|
||||
git_url = _resolve_git_url(identifier)
|
||||
git_url, _subdir = _resolve_git_url(identifier)
|
||||
except ValueError as e:
|
||||
console.print(f"[red]Error:[/red] {e}")
|
||||
sys.exit(1)
|
||||
@@ -482,7 +545,10 @@ def cmd_install(
|
||||
"Consider using https:// or git@ for production installs.",
|
||||
)
|
||||
|
||||
console.print(f"[dim]Cloning {git_url}...[/dim]")
|
||||
if _subdir:
|
||||
console.print(f"[dim]Cloning {git_url} (subdir: {_subdir})...[/dim]")
|
||||
else:
|
||||
console.print(f"[dim]Cloning {git_url}...[/dim]")
|
||||
|
||||
try:
|
||||
target, installed_manifest, installed_name = _install_plugin_core(
|
||||
@@ -1473,7 +1539,7 @@ def dashboard_install_plugin(
|
||||
"""Non-interactive install for the web dashboard. Returns a JSON-serializable dict."""
|
||||
warnings: list[str] = []
|
||||
try:
|
||||
git_url = _resolve_git_url(identifier)
|
||||
git_url, _subdir = _resolve_git_url(identifier)
|
||||
if git_url.startswith(("http://", "file://")):
|
||||
warnings.append(
|
||||
"Insecure URL scheme; prefer https:// or git@ for production installs.",
|
||||
|
||||
@@ -3,6 +3,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@@ -16,6 +18,7 @@ from hermes_cli.plugins_cmd import (
|
||||
_repo_name_from_url,
|
||||
_resolve_git_executable,
|
||||
_resolve_git_url,
|
||||
_resolve_subdir_within,
|
||||
_sanitize_plugin_name,
|
||||
)
|
||||
|
||||
@@ -97,35 +100,127 @@ class TestSanitizePluginName:
|
||||
|
||||
|
||||
class TestResolveGitUrl:
|
||||
"""Shorthand and full-URL resolution."""
|
||||
"""Shorthand and full-URL resolution, with optional subdirectory."""
|
||||
|
||||
def test_owner_repo_shorthand(self):
|
||||
url = _resolve_git_url("owner/repo")
|
||||
url, subdir = _resolve_git_url("owner/repo")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir is None
|
||||
|
||||
def test_https_url_passthrough(self):
|
||||
url = _resolve_git_url("https://github.com/x/y.git")
|
||||
url, subdir = _resolve_git_url("https://github.com/x/y.git")
|
||||
assert url == "https://github.com/x/y.git"
|
||||
assert subdir is None
|
||||
|
||||
def test_ssh_url_passthrough(self):
|
||||
url = _resolve_git_url("git@github.com:x/y.git")
|
||||
url, subdir = _resolve_git_url("git@github.com:x/y.git")
|
||||
assert url == "git@github.com:x/y.git"
|
||||
assert subdir is None
|
||||
|
||||
def test_http_url_passthrough(self):
|
||||
url = _resolve_git_url("http://example.com/repo.git")
|
||||
url, subdir = _resolve_git_url("http://example.com/repo.git")
|
||||
assert url == "http://example.com/repo.git"
|
||||
assert subdir is None
|
||||
|
||||
def test_file_url_passthrough(self):
|
||||
url = _resolve_git_url("file:///tmp/repo")
|
||||
url, subdir = _resolve_git_url("file:///tmp/repo")
|
||||
assert url == "file:///tmp/repo"
|
||||
assert subdir is None
|
||||
|
||||
def test_invalid_single_word_raises(self):
|
||||
with pytest.raises(ValueError, match="Invalid plugin identifier"):
|
||||
_resolve_git_url("justoneword")
|
||||
|
||||
def test_invalid_three_parts_raises(self):
|
||||
with pytest.raises(ValueError, match="Invalid plugin identifier"):
|
||||
_resolve_git_url("a/b/c")
|
||||
def test_shorthand_with_subdir(self):
|
||||
url, subdir = _resolve_git_url("owner/repo/my-plugin")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "my-plugin"
|
||||
|
||||
def test_shorthand_with_nested_subdir(self):
|
||||
url, subdir = _resolve_git_url("owner/repo/path/to/plugin")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "path/to/plugin"
|
||||
|
||||
def test_shorthand_with_subdir_trailing_slash(self):
|
||||
url, subdir = _resolve_git_url("owner/repo/my-plugin/")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "my-plugin"
|
||||
|
||||
def test_https_url_with_subdir(self):
|
||||
url, subdir = _resolve_git_url("https://github.com/owner/repo.git/my-plugin")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "my-plugin"
|
||||
|
||||
def test_https_url_with_nested_subdir(self):
|
||||
url, subdir = _resolve_git_url(
|
||||
"https://github.com/owner/repo.git/path/to/plugin"
|
||||
)
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "path/to/plugin"
|
||||
|
||||
def test_url_with_fragment_subdir(self):
|
||||
url, subdir = _resolve_git_url("https://github.com/owner/repo.git#my-plugin")
|
||||
assert url == "https://github.com/owner/repo.git"
|
||||
assert subdir == "my-plugin"
|
||||
|
||||
def test_file_url_with_fragment_subdir(self):
|
||||
url, subdir = _resolve_git_url("file:///tmp/repo#path/to/plugin")
|
||||
assert url == "file:///tmp/repo"
|
||||
assert subdir == "path/to/plugin"
|
||||
|
||||
def test_ssh_url_with_fragment_subdir(self):
|
||||
url, subdir = _resolve_git_url("git@github.com:owner/repo.git#sub")
|
||||
assert url == "git@github.com:owner/repo.git"
|
||||
assert subdir == "sub"
|
||||
|
||||
|
||||
# ── _resolve_subdir_within ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestResolveSubdirWithin:
|
||||
"""Subdirectory resolution stays within the clone and rejects traversal."""
|
||||
|
||||
def test_valid_subdir(self, tmp_path):
|
||||
(tmp_path / "my-plugin").mkdir()
|
||||
result = _resolve_subdir_within(tmp_path, "my-plugin")
|
||||
assert result == (tmp_path / "my-plugin").resolve()
|
||||
|
||||
def test_valid_nested_subdir(self, tmp_path):
|
||||
(tmp_path / "a" / "b" / "c").mkdir(parents=True)
|
||||
result = _resolve_subdir_within(tmp_path, "a/b/c")
|
||||
assert result == (tmp_path / "a" / "b" / "c").resolve()
|
||||
|
||||
def test_rejects_dot_dot_escape(self, tmp_path):
|
||||
clone = tmp_path / "clone"
|
||||
clone.mkdir()
|
||||
(tmp_path / "secret").mkdir()
|
||||
with pytest.raises(PluginOperationError, match="escapes the repository"):
|
||||
_resolve_subdir_within(clone, "../secret")
|
||||
|
||||
def test_rejects_absolute_path_escape(self, tmp_path):
|
||||
clone = tmp_path / "clone"
|
||||
clone.mkdir()
|
||||
# An absolute path resolves outside the clone root.
|
||||
with pytest.raises(PluginOperationError, match="escapes the repository"):
|
||||
_resolve_subdir_within(clone, "/etc")
|
||||
|
||||
def test_rejects_symlink_escape(self, tmp_path):
|
||||
clone = tmp_path / "clone"
|
||||
clone.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
(clone / "link").symlink_to(outside)
|
||||
with pytest.raises(PluginOperationError, match="escapes the repository"):
|
||||
_resolve_subdir_within(clone, "link")
|
||||
|
||||
def test_rejects_missing_subdir(self, tmp_path):
|
||||
with pytest.raises(PluginOperationError, match="does not exist"):
|
||||
_resolve_subdir_within(tmp_path, "nope")
|
||||
|
||||
def test_rejects_file_not_dir(self, tmp_path):
|
||||
(tmp_path / "afile").write_text("x")
|
||||
with pytest.raises(PluginOperationError, match="not a directory"):
|
||||
_resolve_subdir_within(tmp_path, "afile")
|
||||
|
||||
|
||||
# ── _resolve_git_executable ─────────────────────────────────────────────────
|
||||
@@ -698,3 +793,90 @@ class TestNoAutoActivation:
|
||||
# The old code had: "Even with default config, check if a plugin registered one"
|
||||
# The fix removes this. Verify it's gone.
|
||||
assert "Even with default config, check if a plugin registered one" not in source
|
||||
|
||||
|
||||
# ── End-to-end subdirectory install ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSubdirInstallE2E:
|
||||
"""Install a plugin that lives in a subdirectory of a real local git repo."""
|
||||
|
||||
@staticmethod
|
||||
def _make_repo_with_subdir_plugin(repo_root: Path) -> None:
|
||||
"""Create a git repo where the plugin lives in ``./my-plugin/`` and the
|
||||
repo root holds unrelated docs/tests."""
|
||||
import subprocess as sp
|
||||
|
||||
repo_root.mkdir(parents=True, exist_ok=True)
|
||||
# Root-level noise: docs + tests that should NOT be installed.
|
||||
(repo_root / "README.md").write_text("# Monorepo docs\n")
|
||||
(repo_root / "tests").mkdir()
|
||||
(repo_root / "tests" / "test_x.py").write_text("def test_x():\n pass\n")
|
||||
# The actual plugin in a subdirectory.
|
||||
plugin_dir = repo_root / "my-plugin"
|
||||
plugin_dir.mkdir()
|
||||
(plugin_dir / "plugin.yaml").write_text(
|
||||
"name: my-plugin\nmanifest_version: 1\ndescription: A subdir plugin\n"
|
||||
)
|
||||
(plugin_dir / "__init__.py").write_text("# plugin entry\n")
|
||||
|
||||
env = {
|
||||
**os.environ,
|
||||
"GIT_AUTHOR_NAME": "t",
|
||||
"GIT_AUTHOR_EMAIL": "t@t",
|
||||
"GIT_COMMITTER_NAME": "t",
|
||||
"GIT_COMMITTER_EMAIL": "t@t",
|
||||
}
|
||||
sp.run(["git", "init", "-q"], cwd=repo_root, check=True, env=env)
|
||||
sp.run(["git", "add", "-A"], cwd=repo_root, check=True, env=env)
|
||||
sp.run(
|
||||
["git", "commit", "-q", "-m", "init"],
|
||||
cwd=repo_root,
|
||||
check=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
def test_installs_only_the_subdir_plugin(self, tmp_path, monkeypatch):
|
||||
if shutil.which("git") is None:
|
||||
pytest.skip("git not available")
|
||||
|
||||
from hermes_cli import plugins_cmd as pc
|
||||
|
||||
repo_root = tmp_path / "monorepo"
|
||||
self._make_repo_with_subdir_plugin(repo_root)
|
||||
|
||||
plugins_dir = tmp_path / "installed"
|
||||
plugins_dir.mkdir()
|
||||
monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
|
||||
|
||||
identifier = f"file://{repo_root}#my-plugin"
|
||||
target, manifest, name = pc._install_plugin_core(identifier, force=False)
|
||||
|
||||
# Installed under the plugin's own name, not the repo name.
|
||||
assert name == "my-plugin"
|
||||
assert manifest.get("name") == "my-plugin"
|
||||
assert target == (plugins_dir / "my-plugin").resolve()
|
||||
|
||||
# The plugin's files are present...
|
||||
assert (target / "plugin.yaml").exists()
|
||||
assert (target / "__init__.py").exists()
|
||||
# ...and the repo-root noise is NOT.
|
||||
assert not (target / "README.md").exists()
|
||||
assert not (target / "tests").exists()
|
||||
|
||||
def test_missing_subdir_raises(self, tmp_path, monkeypatch):
|
||||
if shutil.which("git") is None:
|
||||
pytest.skip("git not available")
|
||||
|
||||
from hermes_cli import plugins_cmd as pc
|
||||
|
||||
repo_root = tmp_path / "monorepo"
|
||||
self._make_repo_with_subdir_plugin(repo_root)
|
||||
|
||||
plugins_dir = tmp_path / "installed"
|
||||
plugins_dir.mkdir()
|
||||
monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
|
||||
|
||||
identifier = f"file://{repo_root}#does-not-exist"
|
||||
with pytest.raises(PluginOperationError, match="does not exist"):
|
||||
pc._install_plugin_core(identifier, force=False)
|
||||
|
||||
@@ -362,7 +362,7 @@ export const en: Translations = {
|
||||
inactive: "inactive",
|
||||
installBtn: "Install",
|
||||
installHeading: "Install from GitHub / Git URL",
|
||||
installHint: "Use owner/repo shorthand or a full https:// or git@ clone URL.",
|
||||
installHint: "Use owner/repo shorthand or a full https:// or git@ clone URL. For a plugin in a subdirectory, append the path: owner/repo/path/to/plugin (or <url>#path/to/plugin).",
|
||||
memoryProviderLabel: "Memory provider",
|
||||
missingEnvWarn: "Set these in Keys before the plugin can run:",
|
||||
noDashboardTab: "No dashboard tab",
|
||||
|
||||
@@ -240,7 +240,7 @@ export default function PluginsPage() {
|
||||
<Input
|
||||
className="font-mono-ui lowercase"
|
||||
id="install-url"
|
||||
placeholder="owner/repo or https://..."
|
||||
placeholder="owner/repo, owner/repo/subdir, or https://..."
|
||||
spellCheck={false}
|
||||
value={installId}
|
||||
onChange={(e) => setInstallId(e.target.value)}
|
||||
|
||||
Reference in New Issue
Block a user