diff --git a/agent/secret_sources/registry.py b/agent/secret_sources/registry.py index 216db858baa2..5106d7a1ad73 100644 --- a/agent/secret_sources/registry.py +++ b/agent/secret_sources/registry.py @@ -236,7 +236,10 @@ def _fetch() -> FetchResult: return res except Exception as exc: # noqa: BLE001 — contract violation, contain it res = FetchResult() - res.error = f"fetch raised {type(exc).__name__}: {exc}" + # Do not interpolate a third-party exception. Secret-manager SDKs + # and helper wrappers sometimes include the rejected value in an + # exception message; this error is printed during startup. + res.error = f"fetch raised {type(exc).__name__}" res.error_kind = ErrorKind.INTERNAL return res finally: diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 0f0a98c4f00c..a492d6582c7c 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -198,6 +198,8 @@ def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]: if not cfg: return {} + _discover_configured_secret_source_plugins(home, cfg) + try: from agent.secret_scope import _is_global_env, load_env_file from agent.secret_sources.registry import apply_all @@ -642,6 +644,8 @@ def _apply_external_secret_sources(home_path: Path) -> None: except ImportError: return + _discover_configured_secret_source_plugins(home_path, cfg) + try: report = apply_all(cfg, home_path) except Exception: # noqa: BLE001 — belt-and-braces; apply_all shouldn't raise @@ -694,6 +698,50 @@ def _apply_external_secret_sources(home_path: Path) -> None: print(f" Secret sources: {conflict}", file=sys.stderr) +def _discover_configured_secret_source_plugins( + home_path: Path, + secrets_cfg: dict, +) -> None: + """Register configured plugin sources before ``apply_all`` validates them. + + Dotenv loading precedes ordinary plugin discovery in several entrypoints. + The optional ``secrets.sources`` list and source-specific config sections + can both select an installed plugin source. Only unknown configured source + names trigger a restricted scan of enabled native plugins beneath the + explicit ``home_path``. + + This helper only discovers/registers plugins; fetching remains exclusively + in the subsequent ``apply_all`` call. Discovery failures are deliberately + swallowed without interpolating the exception: startup is fail-open and an + exception raised by third-party code may contain credential material. + """ + if not isinstance(secrets_cfg, dict): + return + + configured = { + name + for name, value in secrets_cfg.items() + if isinstance(name, str) and isinstance(value, dict) + } + explicit = secrets_cfg.get("sources") + if isinstance(explicit, list): + configured.update(name for name in explicit if isinstance(name, str)) + if not configured: + return + + try: + from agent.secret_sources.registry import get_source + + unknown = {name for name in configured if get_source(name) is None} + if not unknown: + return + from hermes_cli.plugins import discover_configured_secret_source_plugins + + discover_configured_secret_source_plugins(home_path, unknown) + except Exception: # noqa: BLE001 — plugin discovery must not block startup + return + + def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict) -> str: """Ask the failed source for its one-line fix-it hint. diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index c6ed42d726e7..9cfbad94aacb 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -33,6 +33,7 @@ from __future__ import annotations +import ast import asyncio import hashlib import importlib.metadata @@ -315,6 +316,8 @@ class PluginManifest: requires_env: List[Union[str, Dict[str, Any]]] = field(default_factory=list) provides_tools: List[str] = field(default_factory=list) provides_hooks: List[str] = field(default_factory=list) + # Optional declaration used by the pre-dotenv source-only discovery phase. + provides_secret_sources: List[str] = field(default_factory=list) source: str = "" # "user", "project", or "entrypoint" path: Optional[str] = None # Plugin kind — see plugins.py module docstring for semantics. @@ -361,6 +364,31 @@ class LoadedPlugin: deferred: bool = False +class _SecretSourceBootstrapContext: + """Restricted plugin context used before normal plugin discovery.""" + + def __init__(self, manifest: PluginManifest, allowed_source_names: Set[str]): + self.manifest = manifest + self._allowed_source_names = frozenset(allowed_source_names) + + def register_secret_source(self, source) -> None: + from agent.secret_sources.base import SecretSource + from agent.secret_sources.registry import register_source + + if ( + isinstance(source, SecretSource) + and source.name in self._allowed_source_names + ): + register_source(source) + + def __getattr__(self, name: str): + # All non-secret capabilities wait for the ordinary discovery phase, + # which imports the plugin again with the complete PluginContext. + if name.startswith("register_"): + return lambda *args, **kwargs: None + raise AttributeError(name) + + # --------------------------------------------------------------------------- # PluginContext – handed to each plugin's ``register()`` function # --------------------------------------------------------------------------- @@ -859,15 +887,10 @@ def register_secret_source(self, source) -> None: ordering, mapped-vs-bulk precedence, conflict warnings, and provenance; the source only fetches. - NOTE ON TIMING: plugin discovery happens later in startup than - the first ``load_hermes_dotenv()`` call, so a plugin-registered - source is not consulted by the initial env load of the process - that discovers it. It IS consulted by every subsequently - spawned Hermes process (gateway children, cron sessions, - subagents), and immediately after a - ``reset_secret_source_cache()`` re-pull. Plugin sources are - therefore best for supplying credentials to the running fleet; - the bundled sources cover first-process bootstrap. + NOTE ON TIMING: ``load_hermes_dotenv()`` performs idempotent plugin + discovery when ``secrets.sources`` contains an as-yet-unregistered + name. The source is therefore available on the initial env load, + before registry validation and credential reads. Contract requirements (rejected with a warning otherwise): inherit from ``SecretSource``, ``api_version`` matching @@ -877,7 +900,7 @@ def register_secret_source(self, source) -> None: See the base-module docstring for the full contract. """ from agent.secret_sources.base import SecretSource - from agent.secret_sources.registry import register_source + from agent.secret_sources.registry import get_source, register_source if not isinstance(source, SecretSource): logger.warning( @@ -886,6 +909,19 @@ def register_secret_source(self, source) -> None: self.manifest.name, ) return + plugin_path = ( + str(Path(self.manifest.path).resolve()) if self.manifest.path else "" + ) + bootstrap_sources = self._manager._secret_source_bootstrap_paths.get( + plugin_path, set() + ) + if source.name in bootstrap_sources and get_source(source.name) is not None: + logger.debug( + "Plugin '%s' retained bootstrap secret source: %s", + self.manifest.name, + source.name, + ) + return if register_source(source): logger.info( "Plugin '%s' registered secret source: %s", @@ -1319,6 +1355,9 @@ def __init__(self) -> None: self._context_engine = None # Set by a plugin via register_context_engine() self._plugin_commands: Dict[str, dict] = {} # Slash commands registered by plugins self._discovered: bool = False + # Restricted pre-dotenv imports are tracked independently: they must + # never make ordinary plugin discovery look complete. + self._secret_source_bootstrap_paths: Dict[str, Set[str]] = {} self._cli_ref = None # Set by CLI after plugin discovery # Plugin skill registry: qualified name → metadata dict. self._plugin_skills: Dict[str, Dict[str, Any]] = {} @@ -1544,6 +1583,170 @@ def _collect_directory_manifests(self) -> List[PluginManifest]: return manifests + def discover_configured_secret_sources( + self, + home_path: Path, + configured_source_names: Set[str], + ) -> None: + """Import only enabled source plugins from one explicit Hermes home. + + This restricted first phase scans only ``/plugins`` and + reads ``plugins.enabled`` from that same home's config. It does not scan + bundled, project, or pip plugins and does not set ``_discovered``. + Normal discovery later imports plugins with the complete context. + """ + if not configured_source_names or _env_enabled("HERMES_SAFE_MODE"): + return + + home_path = Path(home_path) + try: + raw_config = fast_safe_load( + (home_path / "config.yaml").read_text(encoding="utf-8") + ) or {} + except Exception: + return + if not isinstance(raw_config, dict): + return + plugins_config = raw_config.get("plugins") + if not isinstance(plugins_config, dict): + return + enabled_value = plugins_config.get("enabled") + if not isinstance(enabled_value, list): + return + enabled = {value for value in enabled_value if isinstance(value, str)} + disabled_value = plugins_config.get("disabled", []) + disabled = ( + {value for value in disabled_value if isinstance(value, str)} + if isinstance(disabled_value, list) + else set() + ) + if not enabled: + return + + from agent.secret_sources.registry import get_source + + manifests = self._scan_directory( + home_path / "plugins", + source="user", + redact_errors=True, + ) + winners: Dict[str, PluginManifest] = {} + for manifest in manifests: + winners[manifest.key or manifest.name] = manifest + + for manifest in winners.values(): + remaining_sources = { + name for name in configured_source_names if get_source(name) is None + } + if not remaining_sources: + break + lookup_key = manifest.key or manifest.name + if lookup_key in disabled or manifest.name in disabled: + continue + if lookup_key not in enabled and manifest.name not in enabled: + continue + if manifest.portable or not manifest.path: + continue + + declared = set(manifest.provides_secret_sources) + if declared: + if declared.isdisjoint(remaining_sources): + continue + elif not self._module_registers_secret_source(manifest): + # Legacy source plugins may predate manifest declarations. + # Inspect syntax rather than importing unrelated code as a probe. + continue + + resolved_path = str(Path(manifest.path).resolve()) + if resolved_path in self._secret_source_bootstrap_paths: + continue + self._load_secret_source_bootstrap_plugin( + manifest, + remaining_sources, + ) + + @staticmethod + def _module_registers_secret_source(manifest: PluginManifest) -> bool: + """Return whether the plugin entry module calls the source hook.""" + try: + source = (Path(manifest.path) / "__init__.py").read_text( + encoding="utf-8" + ) + tree = ast.parse(source) + except Exception: + return False + return any( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "register_secret_source" + for node in ast.walk(tree) + ) + + def _load_secret_source_bootstrap_plugin( + self, + manifest: PluginManifest, + configured_source_names: Set[str], + ) -> None: + """Run one source plugin with the restricted bootstrap context.""" + from agent.secret_sources.registry import get_source + + plugin_path = str(Path(manifest.path).resolve()) + key = manifest.key or manifest.name + slug = key.replace("/", "__").replace("-", "_") + token = hashlib.sha256( + f"{plugin_path}:{id(self)}".encode() + ).hexdigest()[:16] + module_name = f"{_NS_PARENT}.{slug}__secret_bootstrap_{token}" + canonical_prefix = f"{_NS_PARENT}.{slug}" + canonical_modules = { + name: module + for name, module in sys.modules.items() + if name == canonical_prefix or name.startswith(canonical_prefix + ".") + } + before = { + name for name in configured_source_names if get_source(name) is not None + } + try: + module = self._load_directory_module_as(manifest, module_name) + register_fn = getattr(module, "register", None) + if register_fn is None: + return + register_fn( + _SecretSourceBootstrapContext(manifest, configured_source_names) + ) + except Exception: + # Never interpolate third-party exception text: secret-manager SDKs + # may include rejected credential values in their errors. + logger.warning( + "Enabled secret-source plugin '%s' failed during bootstrap", + manifest.name, + ) + finally: + # Bootstrap must not populate the canonical plugin package cache. + # Package plugins often re-export register() from a submodule; if + # that submodule survives, ordinary discovery can silently skip + # its non-secret capabilities. Restore the exact pre-bootstrap + # canonical cache and discard the temporary package tree. + for name in list(sys.modules): + if name == module_name or name.startswith(module_name + "."): + sys.modules.pop(name, None) + elif ( + name == canonical_prefix + or name.startswith(canonical_prefix + ".") + ) and name not in canonical_modules: + sys.modules.pop(name, None) + sys.modules.update(canonical_modules) + + registered = { + name + for name in configured_source_names + if name not in before and get_source(name) is not None + } + if registered: + self._secret_source_bootstrap_paths.setdefault(plugin_path, set()).update( + registered + ) + def has_enabled_portable_mcp(self, raw_config: Mapping[str, Any]) -> bool: """Probe enabled portable MCP packages without loading plugins. @@ -1609,6 +1812,7 @@ def _scan_directory( path: Path, source: str, skip_names: Optional[Set[str]] = None, + redact_errors: bool = False, ) -> List[PluginManifest]: """Read ``plugin.yaml`` manifests from subdirectories of *path*. @@ -1626,7 +1830,12 @@ def _scan_directory( pass it now that categories are first-class). """ return self._scan_directory_level( - path, source, skip_names=skip_names, prefix="", depth=0 + path, + source, + skip_names=skip_names, + prefix="", + depth=0, + redact_errors=redact_errors, ) def _scan_directory_level( @@ -1637,6 +1846,7 @@ def _scan_directory_level( skip_names: Optional[Set[str]], prefix: str, depth: int, + redact_errors: bool, ) -> List[PluginManifest]: """Recursive implementation of :meth:`_scan_directory`. @@ -1659,7 +1869,11 @@ def _scan_directory_level( if manifest_file.exists(): manifest = self._parse_manifest( - manifest_file, child, source, prefix + manifest_file, + child, + source, + prefix, + redact_errors=redact_errors, ) if manifest is not None: manifests.append(manifest) @@ -1671,12 +1885,13 @@ def _scan_directory_level( from hermes_cli.agent_plugins import read_agent_plugin_manifest data, diagnostics = read_agent_plugin_manifest(child) - for diagnostic in diagnostics: - logger.warning( - "Agent Plugin '%s': %s", - child, - diagnostic.message, - ) + if not redact_errors: + for diagnostic in diagnostics: + logger.warning( + "Agent Plugin '%s': %s", + child, + diagnostic.message, + ) key = f"{prefix}/{child.name}" if prefix else data["name"] manifests.append( PluginManifest( @@ -1692,7 +1907,12 @@ def _scan_directory_level( ) ) except Exception as exc: - logger.warning("Failed to parse %s: %s", portable_file, exc) + if redact_errors: + logger.warning( + "Failed to parse plugin manifest during secret bootstrap" + ) + else: + logger.warning("Failed to parse %s: %s", portable_file, exc) continue # No manifest at this level. If we're still within the depth @@ -1710,6 +1930,7 @@ def _scan_directory_level( skip_names=None, prefix=sub_prefix, depth=depth + 1, + redact_errors=redact_errors, ) ) @@ -1721,6 +1942,8 @@ def _parse_manifest( plugin_dir: Path, source: str, prefix: str, + *, + redact_errors: bool = False, ) -> Optional[PluginManifest]: """Parse a single ``plugin.yaml`` into a :class:`PluginManifest`. @@ -1795,15 +2018,30 @@ def _parse_manifest( requires_env=data.get("requires_env", []), provides_tools=data.get("provides_tools", []), provides_hooks=data.get("provides_hooks", []), + provides_secret_sources=[ + value + for value in data.get("provides_secret_sources", []) + if isinstance(value, str) + ] + if isinstance(data.get("provides_secret_sources", []), list) + else [], source=source, path=str(plugin_dir), kind=kind, key=key, ) except Exception as exc: - logger.warning( - "Failed to parse %s: %s", manifest_file, exc, exc_info=_PLUGINS_DEBUG, - ) + if redact_errors: + logger.warning( + "Failed to parse plugin manifest during secret bootstrap" + ) + else: + logger.warning( + "Failed to parse %s: %s", + manifest_file, + exc, + exc_info=_PLUGINS_DEBUG, + ) return None # ----------------------------------------------------------------------- @@ -2048,21 +2286,28 @@ def _load_directory_module(self, manifest: PluginManifest) -> types.ModuleType: ``hermes_plugins.image_gen__openai`` without colliding with any future ``tts/openai``. """ + key = manifest.key or manifest.name + slug = key.replace("/", "__").replace("-", "_") + module_name = f"{_NS_PARENT}.{slug}" + return self._load_directory_module_as(manifest, module_name) + + def _load_directory_module_as( + self, + manifest: PluginManifest, + module_name: str, + ) -> types.ModuleType: + """Import one directory plugin under an explicit package name.""" plugin_dir = Path(manifest.path) # type: ignore[arg-type] init_file = plugin_dir / "__init__.py" if not init_file.exists(): raise FileNotFoundError(f"No __init__.py in {plugin_dir}") - # Ensure the namespace parent package exists if _NS_PARENT not in sys.modules: ns_pkg = types.ModuleType(_NS_PARENT) ns_pkg.__path__ = [] # type: ignore[attr-defined] ns_pkg.__package__ = _NS_PARENT sys.modules[_NS_PARENT] = ns_pkg - key = manifest.key or manifest.name - slug = key.replace("/", "__").replace("-", "_") - module_name = f"{_NS_PARENT}.{slug}" spec = importlib.util.spec_from_file_location( module_name, init_file, @@ -2281,6 +2526,17 @@ def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool: return PluginManager().has_enabled_portable_mcp(raw_config) +def discover_configured_secret_source_plugins( + home_path: Path, + configured_source_names: Set[str], +) -> None: + """Run restricted pre-dotenv discovery for one explicit home.""" + get_plugin_manager().discover_configured_secret_sources( + home_path, + configured_source_names, + ) + + def discover_plugins(force: bool = False) -> None: """Discover and load all plugins. diff --git a/tests/test_external_secret_source_plugin_startup.py b/tests/test_external_secret_source_plugin_startup.py new file mode 100644 index 000000000000..8a60163d5472 --- /dev/null +++ b/tests/test_external_secret_source_plugin_startup.py @@ -0,0 +1,397 @@ +"""First-load regressions for plugin-provided external secret sources.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def _write_secret_source_plugin( + home: Path, + *, + fetch_body: str, + source_name: str = "startup_test_source", + plugin_name: str = "startup-test-plugin", + declares_source: bool = True, + register_body: str | None = None, +) -> None: + plugin_dir = home / "plugins" / plugin_name + plugin_dir.mkdir(parents=True) + declaration = ( + f"provides_secret_sources:\n - {source_name}\n" if declares_source else "" + ) + (plugin_dir / "plugin.yaml").write_text( + f"name: {plugin_name}\n" + "kind: standalone\n" + "version: 1.0.0\n" + "description: External secret-source startup test\n" + f"{declaration}", + encoding="utf-8", + ) + register = register_body or ( + "def register(ctx):\n" + " ctx.register_secret_source(StartupTestSource())\n" + ) + (plugin_dir / "__init__.py").write_text( + "from agent.secret_sources.base import FetchResult, SecretSource\n" + "\n" + "class StartupTestSource(SecretSource):\n" + f" name = {source_name!r}\n" + " label = 'Startup test source'\n" + " shape = 'mapped'\n" + "\n" + " def fetch(self, cfg, home_path):\n" + f"{fetch_body}\n" + "\n" + f"{register}", + encoding="utf-8", + ) + + +def _write_package_secret_source_plugin(home: Path, import_marker: Path) -> None: + plugin_dir = home / "plugins" / "startup-package-plugin" + plugin_dir.mkdir(parents=True) + (plugin_dir / "plugin.yaml").write_text( + "name: startup-package-plugin\n" + "kind: standalone\n" + "version: 1.0.0\n" + "description: Package secret-source startup test\n" + "provides_secret_sources:\n" + " - startup_package_source\n", + encoding="utf-8", + ) + (plugin_dir / "__init__.py").write_text( + "from .registration import register\n", + encoding="utf-8", + ) + (plugin_dir / "source.py").write_text( + "from agent.secret_sources.base import FetchResult, SecretSource\n" + "class PackageSource(SecretSource):\n" + " name = 'startup_package_source'\n" + " label = 'Package startup source'\n" + " shape = 'mapped'\n" + " def fetch(self, cfg, home_path):\n" + " return FetchResult(secrets={'STARTUP_TEST_API_KEY': 'package-value'})\n", + encoding="utf-8", + ) + (plugin_dir / "registration.py").write_text( + "from pathlib import Path\n" + "from .source import PackageSource\n" + f"_marker = Path({str(import_marker)!r})\n" + "_prior = _marker.read_text() if _marker.exists() else ''\n" + "_marker.write_text(_prior + 'import\\n')\n" + "_registered = False\n" + "def _command(_args):\n" + " return 'package-command-ok'\n" + "def register(ctx):\n" + " global _registered\n" + " if _registered:\n" + " return\n" + " _registered = True\n" + " ctx.register_secret_source(PackageSource())\n" + " ctx.register_command('startup-package-command', _command, description='package command')\n", + encoding="utf-8", + ) + + +def _write_unrelated_plugin(home: Path, marker: Path) -> None: + plugin_dir = home / "plugins" / "unrelated-plugin" + plugin_dir.mkdir(parents=True) + (plugin_dir / "plugin.yaml").write_text( + "name: unrelated-plugin\n" + "kind: standalone\n" + "version: 1.0.0\n" + "description: Must not import during secret bootstrap\n", + encoding="utf-8", + ) + (plugin_dir / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('imported')\n" + "def register(ctx):\n" + " return None\n", + encoding="utf-8", + ) + + +def _write_config( + home: Path, + *, + source_name: str = "startup_test_source", + plugin_name: str = "startup-test-plugin", + include_sources: bool = True, + include_unrelated: bool = False, +) -> None: + enabled = [plugin_name] + if include_unrelated: + enabled.append("unrelated-plugin") + sources = f" sources:\n - {source_name}\n" if include_sources else "" + (home / "config.yaml").write_text( + "plugins:\n" + " enabled:\n" + + "".join(f" - {name}\n" for name in enabled) + + "secrets:\n" + + sources + + f" {source_name}:\n" + " enabled: true\n" + " env:\n" + " STARTUP_TEST_API_KEY: test-ref\n", + encoding="utf-8", + ) + + +def _run_python( + cwd: Path, + code: str, + *, + process_home: Path | None = None, +) -> subprocess.CompletedProcess[str]: + process_home = process_home or cwd + bundled = process_home / "empty-bundled-plugins" + bundled.mkdir(parents=True, exist_ok=True) + env = os.environ.copy() + env.update( + { + "HERMES_HOME": str(process_home), + "HERMES_BUNDLED_PLUGINS": str(bundled), + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONPATH": os.pathsep.join( + part for part in (str(ROOT), env.get("PYTHONPATH", "")) if part + ), + } + ) + env.pop("HERMES_SAFE_MODE", None) + env.pop("STARTUP_TEST_API_KEY", None) + return subprocess.run( + [sys.executable, "-c", code], + cwd=cwd, + env=env, + text=True, + capture_output=True, + timeout=30, + check=False, + ) + + +def test_first_load_imports_only_enabled_secret_source_plugins(tmp_path): + secret_value = "startup-value-that-must-not-be-logged" + fetch_marker = tmp_path / "fetch-count" + unrelated_marker = tmp_path / "unrelated-imported" + _write_secret_source_plugin( + tmp_path, + fetch_body=( + " marker = home_path / 'fetch-count'\n" + " prior = marker.read_text() if marker.exists() else ''\n" + " marker.write_text(prior + 'fetch\\n')\n" + f" return FetchResult(secrets={{'STARTUP_TEST_API_KEY': {secret_value!r}}})" + ), + ) + _write_unrelated_plugin(tmp_path, unrelated_marker) + _write_config(tmp_path, include_unrelated=True) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os\n" + "from hermes_cli.env_loader import get_secret_source, load_hermes_dotenv\n" + "from hermes_cli.plugins import get_plugin_manager\n" + "home = Path(os.environ['HERMES_HOME'])\n" + "load_hermes_dotenv(hermes_home=home)\n" + "load_hermes_dotenv(hermes_home=home)\n" + f"assert os.environ['STARTUP_TEST_API_KEY'] == {secret_value!r}\n" + "assert get_secret_source('STARTUP_TEST_API_KEY') == 'startup_test_source'\n" + "assert not get_plugin_manager()._discovered\n" + "print('targeted-bootstrap-ok')\n", + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "targeted-bootstrap-ok" + assert fetch_marker.read_text() == "fetch\n" + assert not unrelated_marker.exists() + assert "unknown source" not in (result.stdout + result.stderr).lower() + assert secret_value not in result.stdout + assert secret_value not in result.stderr + + +def test_package_plugin_registers_non_secret_capabilities_after_bootstrap(tmp_path): + import_marker = tmp_path / "package-import-count" + _write_package_secret_source_plugin(tmp_path, import_marker) + _write_config( + tmp_path, + source_name="startup_package_source", + plugin_name="startup-package-plugin", + ) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os, sys\n" + "from hermes_cli.env_loader import load_hermes_dotenv\n" + "from hermes_cli.plugins import get_plugin_commands, get_plugin_manager\n" + "home = Path(os.environ['HERMES_HOME'])\n" + "load_hermes_dotenv(hermes_home=home)\n" + "manager = get_plugin_manager()\n" + "assert not manager._discovered\n" + "assert not any('__secret_bootstrap_' in name for name in sys.modules)\n" + "commands = get_plugin_commands()\n" + "assert commands['startup-package-command']['handler']('') == 'package-command-ok'\n" + "loaded = manager._plugins['startup-package-plugin']\n" + "assert loaded.enabled and loaded.error is None\n" + "assert 'startup-package-command' in loaded.commands_registered\n" + "assert not any('__secret_bootstrap_' in name for name in sys.modules)\n" + "print('package-full-discovery-ok')\n", + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "package-full-discovery-ok" + assert import_marker.read_text() == "import\nimport\n" + assert "already registered" not in result.stderr + + +def test_explicit_non_process_home_is_honored(tmp_path): + process_home = tmp_path / "process-home" + target_home = tmp_path / "target-home" + process_home.mkdir() + target_home.mkdir() + _write_secret_source_plugin( + target_home, + fetch_body=( + " (home_path / 'target-fetch').write_text(str(home_path))\n" + " return FetchResult(secrets={'STARTUP_TEST_API_KEY': 'target-value'})" + ), + ) + _write_config(target_home) + + result = _run_python( + target_home, + "from pathlib import Path\n" + "import os\n" + "from hermes_cli.env_loader import load_hermes_dotenv\n" + f"target = Path({str(target_home)!r})\n" + "load_hermes_dotenv(hermes_home=target)\n" + "assert os.environ['STARTUP_TEST_API_KEY'] == 'target-value'\n" + "assert (target / 'target-fetch').read_text() == str(target)\n" + "print('explicit-home-ok')\n", + process_home=process_home, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "explicit-home-ok" + assert not (process_home / "target-fetch").exists() + + +def test_omitted_sources_list_uses_enabled_source_section(tmp_path): + _write_secret_source_plugin( + tmp_path, + declares_source=False, + fetch_body=( + " return FetchResult(secrets={'STARTUP_TEST_API_KEY': 'legacy-value'})" + ), + ) + _write_config(tmp_path, include_sources=False) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os\n" + "from hermes_cli.env_loader import load_hermes_dotenv\n" + "home = Path(os.environ['HERMES_HOME'])\n" + "load_hermes_dotenv(hermes_home=home)\n" + "assert os.environ['STARTUP_TEST_API_KEY'] == 'legacy-value'\n" + "print('omitted-list-ok')\n", + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "omitted-list-ok" + + +def test_bootstrap_failure_is_fail_open_and_redacted(tmp_path): + sentinel = "bootstrap-exception-secret-that-must-not-be-logged" + _write_secret_source_plugin( + tmp_path, + fetch_body=" return FetchResult(secrets={})", + register_body=( + "def register(ctx):\n" + f" raise RuntimeError({sentinel!r})\n" + ), + ) + _write_config(tmp_path) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os\n" + "from hermes_cli.env_loader import load_hermes_dotenv\n" + "load_hermes_dotenv(hermes_home=Path(os.environ['HERMES_HOME']))\n" + "assert 'STARTUP_TEST_API_KEY' not in os.environ\n" + "print('bootstrap-survived')\n", + ) + + assert result.returncode == 0 + assert result.stdout.strip() == "bootstrap-survived" + assert sentinel not in result.stdout + assert sentinel not in result.stderr + + +def test_fetch_failure_is_fail_open_and_redacted(tmp_path): + sentinel = "fetch-exception-secret-that-must-not-be-logged" + _write_secret_source_plugin( + tmp_path, + fetch_body=f" raise RuntimeError({sentinel!r})", + ) + _write_config(tmp_path) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os\n" + "from hermes_cli.env_loader import load_hermes_dotenv\n" + "load_hermes_dotenv(hermes_home=Path(os.environ['HERMES_HOME']))\n" + "assert 'STARTUP_TEST_API_KEY' not in os.environ\n" + "print('fetch-survived')\n", + ) + + assert result.returncode == 0 + assert result.stdout.strip() == "fetch-survived" + assert sentinel not in result.stdout + assert sentinel not in result.stderr + + +def test_benign_secret_config_read_does_not_import_or_fetch_plugin(tmp_path): + import_marker = tmp_path / "plugin-imported" + plugin_dir = tmp_path / "plugins" / "startup-test-plugin" + plugin_dir.mkdir(parents=True) + (plugin_dir / "plugin.yaml").write_text( + "name: startup-test-plugin\n" + "kind: standalone\n" + "provides_secret_sources:\n" + " - startup_test_source\n", + encoding="utf-8", + ) + (plugin_dir / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(import_marker)!r}).write_text('imported')\n", + encoding="utf-8", + ) + _write_config(tmp_path) + + result = _run_python( + tmp_path, + "from pathlib import Path\n" + "import os\n" + "from agent.secret_sources.registry import get_source\n" + "from hermes_cli.env_loader import _load_secrets_config\n" + "home = Path(os.environ['HERMES_HOME'])\n" + "cfg = _load_secrets_config(home)\n" + "assert cfg['sources'] == ['startup_test_source']\n" + "assert get_source('startup_test_source') is None\n" + "print('config-read-only')\n", + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "config-read-only" + assert not import_marker.exists() diff --git a/website/docs/developer-guide/secret-source-plugin.md b/website/docs/developer-guide/secret-source-plugin.md index 5df9f803a692..799ce6eb9cb9 100644 --- a/website/docs/developer-guide/secret-source-plugin.md +++ b/website/docs/developer-guide/secret-source-plugin.md @@ -30,10 +30,23 @@ The orchestrator (`agent.secret_sources.registry.apply_all`) owns everything sec ``` ~/.hermes/plugins/my-vault/ -├── plugin.yaml # name, description +├── plugin.yaml # name, description, provides_secret_sources └── __init__.py # SecretSource subclass + register(ctx) ``` +Declare the source name in the manifest so the pre-dotenv bootstrap can select +this plugin without importing unrelated enabled plugins: + +```yaml +name: my-vault +kind: standalone +provides_secret_sources: + - myvault +``` + +Direct `ctx.register_secret_source(...)` calls remain detectable for older +plugins, but new plugins should include the declaration. + ## The SecretSource ABC Implement `agent.secret_sources.base.SecretSource`. One method is required: