ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(plugins): delegate secret-source enablement to is_enabled contract (#64177)

Browse Source 
Address teknium1 review on #64189:
- Re-pull gate now delegates to each source's is_enabled(cfg) via the
  registry contract, so a plugin source with custom activation logic is
  honored (previously only secrets.<name>.enabled was checked).
- Add BUILTIN_SOURCE_NAMES to the registry so plugin-vs-bundled is a
  single source of truth instead of a hard-coded set at the call site.
- Reconcile docs: rewrite the timing :::note to describe both the
  post-discovery re-pull and the remaining import-time limitation, and
  cross-link the first-process bootstrap section.
- Tests: real SecretSource subclasses, custom is_enabled activation
  (positive + negative), is_enabled-raises skip, builtin-only no-op,
  and a discovery-registration end-to-end re-pull check.
This commit is contained in:
Bartok9
2026-07-15 23:17:11 -04:00
committed by Teknium
parent 7a7e73d310
commit 2e29de2296
4 changed files with 217 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/secret_sources/registry.py
+5
View File
@@ -51,6 +51,11 @@ logger = logging.getLogger(__name__)
_SOURCES: Dict[str, SecretSource] = {}
_BUILTINS_LOADED = False
# Canonical names of the deliberately-closed bundled source set. Callers
# (e.g. the plugin-discovery re-pull) use this to distinguish plugin sources
# from in-tree ones without hard-coding a list at each call site.
BUILTIN_SOURCE_NAMES: frozenset = frozenset({"bitwarden", "onepassword"})
@dataclass
class AppliedVar:
hermes_cli/plugins.py
+29 -18
View File
@@ -1379,13 +1379,21 @@ class PluginManager:
raise
def _refresh_secret_sources_after_discovery(self) -> None:
"""If any non-bundled secret source is enabled, reset cache and re-apply.
"""If any plugin secret source is enabled, reset cache and re-apply.
No-op when only built-in sources exist or no secrets config is enabled.
Enablement is delegated to each source's ``is_enabled(cfg)`` — the
same contract the orchestrator uses (``registry._ordered_enabled_sources``)
— so a source with custom activation logic is honored, not just
``secrets.<name>.enabled``.
No-op when only bundled sources exist or none are enabled.
Fail-open: never raise into discover_and_load.
"""
try:
from agent.secret_sources.registry import list_sources
from agent.secret_sources.registry import (
BUILTIN_SOURCE_NAMES,
list_sources,
)
from hermes_cli.env_loader import load_hermes_dotenv, reset_secret_source_cache
except Exception:
return
@@ -1393,13 +1401,13 @@ class PluginManager:
sources = list_sources()
except Exception:
return
builtin = {"bitwarden", "onepassword", "1password"}
plugin_names = [
getattr(s, "name", "") for s in sources if getattr(s, "name", "") not in builtin
plugin_sources = [
s for s in sources if getattr(s, "name", "") not in BUILTIN_SOURCE_NAMES
]
if not plugin_names:
if not plugin_sources:
return
# Only re-pull when at least one plugin source appears enabled in config.
# Load the secrets config once; hand each source its own section and
# let its is_enabled() decide (honours custom activation extensions).
try:
from hermes_cli.config import load_config
@@ -1407,23 +1415,26 @@ class PluginManager:
secrets = cfg.get("secrets") or {}
except Exception:
secrets = {}
enabled_plugin = False
for name in plugin_names:
enabled_names = []
for source in plugin_sources:
name = getattr(source, "name", "")
section = secrets.get(name)
if isinstance(section, dict) and section.get("enabled"):
enabled_plugin = True
break
if section is True:
enabled_plugin = True
break
if not enabled_plugin:
section = section if isinstance(section, dict) else {}
try:
if source.is_enabled(section):
enabled_names.append(name)
except Exception:
# A source whose is_enabled() raises is skipped, mirroring
# the orchestrator's defensive posture.
continue
if not enabled_names:
return
try:
reset_secret_source_cache()
load_hermes_dotenv()
logger.debug(
"Re-applied secret sources after plugin discovery for: %s",
", ".join(sorted(plugin_names)),
", ".join(sorted(enabled_names)),
)
except Exception as exc:
logger.debug("secret source re-apply after discovery failed: %s", exc)
tests/hermes_cli/test_secret_source_bootstrap.py
+182 -35
View File
@@ -2,73 +2,186 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from agent.secret_sources.base import (
SECRET_SOURCE_API_VERSION,
FetchResult,
SecretSource,
)
from hermes_cli.plugins import PluginManager
class _StubSource(SecretSource):
"""Minimal spec-compliant plugin source for tests."""
api_version = SECRET_SOURCE_API_VERSION
shape = "bulk"
def __init__(self, name: str = "myvault", scheme: str | None = None):
self.name = name
self.scheme = scheme
def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
return FetchResult(secrets={})
class _CustomActivationSource(_StubSource):
"""Ignores ``enabled`` and activates when a custom key is present."""
def is_enabled(self, cfg: dict) -> bool:
return bool(isinstance(cfg, dict) and cfg.get("vault_id"))
def test_refresh_secret_sources_noop_without_plugin_sources(monkeypatch):
mgr = PluginManager()
called = {"reset": 0, "load": 0}
def _list():
return []
monkeypatch.setattr(
"agent.secret_sources.registry.list_sources", _list, raising=False
)
import agent.secret_sources.registry as reg
monkeypatch.setattr(reg, "list_sources", _list)
def _boom_reset():
called["reset"] += 1
def _boom_load(**kwargs):
called["load"] += 1
monkeypatch.setattr(reg, "list_sources", lambda: [])
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache", _boom_reset
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
monkeypatch.setattr("hermes_cli.env_loader.load_hermes_dotenv", _boom_load)
mgr._refresh_secret_sources_after_discovery()
assert called["reset"] == 0
assert called["load"] == 0
assert called == {"reset": 0, "load": 0}
def test_refresh_secret_sources_noop_when_only_builtins(monkeypatch):
"""Bundled sources must never trigger a re-pull."""
mgr = PluginManager()
called = {"reset": 0, "load": 0}
import agent.secret_sources.registry as reg
monkeypatch.setattr(
reg, "list_sources", lambda: [_StubSource(name="bitwarden")]
)
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"bitwarden": {"enabled": True}}},
)
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
mgr._refresh_secret_sources_after_discovery()
assert called == {"reset": 0, "load": 0}
def test_refresh_secret_sources_repulls_when_plugin_enabled(monkeypatch):
mgr = PluginManager()
called = {"reset": 0, "load": 0}
class _Src:
name = "myvault"
import agent.secret_sources.registry as reg
monkeypatch.setattr(reg, "list_sources", lambda: [_Src()])
monkeypatch.setattr(reg, "list_sources", lambda: [_StubSource()])
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"myvault": {"enabled": True}}},
)
def _reset():
called["reset"] += 1
def _load(**kwargs):
called["load"] += 1
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache", _reset
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
monkeypatch.setattr("hermes_cli.env_loader.load_hermes_dotenv", _load)
mgr._refresh_secret_sources_after_discovery()
assert called["reset"] == 1
assert called["load"] == 1
assert called == {"reset": 1, "load": 1}
def test_refresh_respects_custom_is_enabled(monkeypatch):
"""A source with custom activation (no ``enabled`` key) is re-pulled."""
mgr = PluginManager()
called = {"reset": 0, "load": 0}
import agent.secret_sources.registry as reg
monkeypatch.setattr(reg, "list_sources", lambda: [_CustomActivationSource()])
# No `enabled` key at all — only the source's custom contract decides.
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"myvault": {"vault_id": "abc123"}}},
)
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
mgr._refresh_secret_sources_after_discovery()
assert called == {"reset": 1, "load": 1}
def test_refresh_skips_custom_source_when_not_activated(monkeypatch):
mgr = PluginManager()
called = {"reset": 0, "load": 0}
import agent.secret_sources.registry as reg
monkeypatch.setattr(reg, "list_sources", lambda: [_CustomActivationSource()])
# `enabled: true` but the custom contract ignores it and requires vault_id.
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"myvault": {"enabled": True}}},
)
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
mgr._refresh_secret_sources_after_discovery()
assert called == {"reset": 0, "load": 0}
def test_refresh_skips_source_whose_is_enabled_raises(monkeypatch):
mgr = PluginManager()
called = {"reset": 0, "load": 0}
class _Boom(_StubSource):
def is_enabled(self, cfg: dict) -> bool:
raise RuntimeError("boom")
import agent.secret_sources.registry as reg
monkeypatch.setattr(reg, "list_sources", lambda: [_Boom()])
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"myvault": {"enabled": True}}},
)
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: called.__setitem__("reset", called["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: called.__setitem__("load", called["load"] + 1),
)
mgr._refresh_secret_sources_after_discovery()
assert called == {"reset": 0, "load": 0}
def test_discover_and_load_invokes_refresh(monkeypatch):
@@ -82,3 +195,37 @@ def test_discover_and_load_invokes_refresh(monkeypatch):
)
mgr.discover_and_load()
assert hits["n"] == 1
def test_real_plugin_source_discovery_applies_dotenv(monkeypatch, tmp_path):
"""End-to-end: a real plugin source registered via discovery triggers a
re-pull that flows through the registry's is_enabled contract."""
import agent.secret_sources.registry as reg
reg._reset_registry_for_tests()
# Register a real plugin source the way PluginContext.register_secret_source
# would (lands in registry.register_source).
plugin_source = _StubSource(name="tmpvault", scheme="tmpvault")
assert reg.register_source(plugin_source) is True
assert any(getattr(s, "name", "") == "tmpvault" for s in reg.list_sources())
applied = {"reset": 0, "load": 0}
monkeypatch.setattr(
"hermes_cli.env_loader.reset_secret_source_cache",
lambda: applied.__setitem__("reset", applied["reset"] + 1),
)
monkeypatch.setattr(
"hermes_cli.env_loader.load_hermes_dotenv",
lambda **kw: applied.__setitem__("load", applied["load"] + 1),
)
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"secrets": {"tmpvault": {"enabled": True}}},
)
mgr = PluginManager()
mgr._refresh_secret_sources_after_discovery()
assert applied == {"reset": 1, "load": 1}
reg._reset_registry_for_tests()
website/docs/developer-guide/secret-source-plugin.md
+1 -1
View File
@@ -141,7 +141,7 @@ def register(ctx):
Registration is rejected (with a log warning, never a crash) for: non-`SecretSource` instances, invalid/duplicate names, a `scheme` another source owns, wrong `api_version`, or a `shape` outside `mapped`/`bulk`.
:::note Timing
Plugin discovery runs later in startup than the first `load_hermes_dotenv()` call, so a plugin source is not consulted by the very first env load of the process that discovers it. It IS consulted by every subsequently spawned Hermes process (gateway children, cron sessions, subagents). Bundled sources cover first-process bootstrap.
Plugin discovery runs later in startup than the first `load_hermes_dotenv()` call. Immediately after discovery, Hermes re-pulls enabled plugin secret sources (`reset_secret_source_cache()` + `load_hermes_dotenv()`), so the discovering process *does* pick them up — see [First-process bootstrap timing](#first-process-bootstrap-timing) above (#64177). The re-pull is fail-open and skipped when no plugin source is enabled. Any code that read `os.environ` *before* discovery completes (i.e. at pure import time) still sees only the initial load; bundled sources remain the safest choice for the earliest bootstrap. Every subsequently spawned Hermes process (gateway children, cron sessions, subagents) also consults plugin sources on its own initial load.
:::
## Users configure it like any other source