diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 8ad9209441..2a04c04edd 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -4052,17 +4052,20 @@ def platform_manifest_stamp(home: Optional[Path] = None) -> tuple: def _platform_plugin_manifests(home: Optional[Path] = None, source: PlatformManifestSource = "all", *, - strict: bool = False): + strict: bool = False, skipped: "list | None" = None): """Yield ``(dir_name, manifest_dict)`` for every platform plugin manifest (see :func:`_platform_manifest_paths`). ``strict`` raises when a manifest cannot be read instead of skipping it: the child-env scrub must not lose a declared secret to an I/O error. Only a manifest known to be a platform's counts (the bundled and ``plugins/platforms/`` dirs); a flat ``plugins/*`` manifest proves it is one only by its content, so an unreadable one is skipped with a warning, as is an unsearchable plugin directory. A manifest that does not - parse declares nothing (its adapter cannot load either) and is skipped.""" + parse declares nothing (its adapter cannot load either) and is skipped. Every skip is appended + to ``skipped``, so a caller can tell a complete scan from a partial one.""" for dir_name, manifest_path, require_kind, st in _platform_manifest_paths(home, source): if manifest_path is None: logger.warning("Skipping unreadable plugin directory %s: %s", dir_name, st) + if skipped is not None: + skipped.append(dir_name) continue try: with open(manifest_path, "r", encoding="utf-8-sig") as f: @@ -4071,8 +4074,12 @@ def _platform_plugin_manifests(home: Optional[Path] = None, source: PlatformMani if strict and not require_kind: raise logger.warning("Skipping unreadable plugin manifest %s: %s", manifest_path, exc) + if skipped is not None: + skipped.append(str(manifest_path)) continue except Exception: + if skipped is not None: + skipped.append(str(manifest_path)) continue if not isinstance(manifest, dict) or (require_kind and manifest.get("kind") != "platform"): continue @@ -4118,6 +4125,15 @@ def platform_manifest_secret_envs(home: Optional[Path] = None, source: PlatformM return _manifest_secret_envs(_platform_plugin_manifests(home, source, strict=strict)) +def platform_manifest_secret_scan(home: Optional[Path] = None) -> "tuple[frozenset[str], bool]": + """``home``'s user-installed platform plugin secrets, strictly read, and whether the scan was + complete: False when a plugin dir or flat manifest could not be read or parsed, so the caller + keeps the denials it already knew instead of releasing them on a failed discovery.""" + skipped: list = [] + names = _manifest_secret_envs(_platform_plugin_manifests(home, "user", strict=True, skipped=skipped)) + return names, not skipped + + def _inject_platform_plugin_env_vars() -> "frozenset[str] | None": """Populate OPTIONAL_ENV_VARS from platform plugin manifests (bundled AND user-installed) so Teams / IRC / Google Chat and third-party platforms are configurable in the ``hermes config`` / diff --git a/tests/tools/test_local_env_blocklist.py b/tests/tools/test_local_env_blocklist.py index d6dde78820..dbc2173fad 100644 --- a/tests/tools/test_local_env_blocklist.py +++ b/tests/tools/test_local_env_blocklist.py @@ -176,6 +176,73 @@ def test_user_platform_plugin_secrets_belong_to_their_own_profile(child_env, mon reset_hermes_home_override(token) +@pytest.mark.skipif(os.name == "nt" or os.geteuid() == 0, # windows-footgun: ok — short-circuits on nt + reason="needs POSIX permissions as non-root") +@pytest.mark.parametrize("layout", ["platforms", "flat"]) +def test_a_failed_rescan_keeps_the_denials_it_already_knew(child_env, monkeypatch, layout): + """healthy -> unreadable -> recovered: a plugin the scan cannot read right now still declared + its secret a moment ago, and the value may already be in the process or profile overlay.""" + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + home = child_env / "profiles" / "a" + if layout == "platforms": + _user_platform_plugin(home, "chatx", "CHATX_SIGNING_SECRET") + plugin_dir = home / "plugins" / "platforms" / "chatx" + locked = plugin_dir # unsearchable dir: skipped, the scan is partial + else: + plugin_dir = home / "plugins" / "chatx" + plugin_dir.mkdir(parents=True) + (plugin_dir / "plugin.yaml").write_text( + "name: chatx\nkind: platform\nrequires_env:\n - name: CHATX_SIGNING_SECRET\n password: true\n", + encoding="utf-8") + locked = plugin_dir / "plugin.yaml" # unreadable flat manifest: skipped, the scan is partial + monkeypatch.setenv("CHATX_SIGNING_SECRET", "fake-value") + token = set_hermes_home_override(home) + try: + def stripped(): + return "CHATX_SIGNING_SECRET" not in local.hermes_subprocess_env(inherit_credentials=True) + assert stripped() + locked.chmod(0) + try: + assert stripped() + assert stripped() # and again, from the uncached partial result + finally: + locked.chmod(0o755 if layout == "platforms" else 0o644) + assert stripped() + finally: + reset_hermes_home_override(token) + + +def test_a_partial_scan_is_rescanned_on_the_next_spawn(child_env, monkeypatch): + """A manifest read that fails once (same file signature afterwards) must not pin the partial + result: the next spawn reads it and strips the secret.""" + import builtins + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + home = child_env / "profiles" / "a" + plugin_dir = home / "plugins" / "chatx" + plugin_dir.mkdir(parents=True) + manifest = plugin_dir / "plugin.yaml" + manifest.write_text( + "name: chatx\nkind: platform\nrequires_env:\n - name: CHATX_SIGNING_SECRET\n password: true\n", + encoding="utf-8") + real_open, failures = builtins.open, [1] + + def flaky_open(path, *args, **kwargs): + if failures and Path(path) == manifest: + failures.pop() + raise PermissionError(13, "denied", str(path)) + return real_open(path, *args, **kwargs) + + monkeypatch.setattr(builtins, "open", flaky_open) + monkeypatch.setenv("CHATX_SIGNING_SECRET", "fake-value") + token = set_hermes_home_override(home) + try: + local.hermes_subprocess_env(inherit_credentials=True) + assert not failures # the failed read happened + assert "CHATX_SIGNING_SECRET" not in local.hermes_subprocess_env(inherit_credentials=True) + finally: + reset_hermes_home_override(token) + + @pytest.mark.platforms("posix") # symlinks def test_a_symlinked_home_alias_shares_its_profiles_plugin_declarations(child_env, monkeypatch): from hermes_constants import reset_hermes_home_override, set_hermes_home_override diff --git a/tools/environments/local_env_policy.py b/tools/environments/local_env_policy.py index 74d2f7927f..bbba55e310 100644 --- a/tools/environments/local_env_policy.py +++ b/tools/environments/local_env_policy.py @@ -124,16 +124,22 @@ _HOME_ADAPTER_SECRET_CACHE: dict[str, tuple] = {} def _home_adapter_secret_env() -> frozenset: """Secrets declared by the bound profile's own user-installed platform plugins. Per home, not process-wide: under multiplex profile A's plugin must neither strip a same-named value from - profile B's children nor be missing from A's. Cached per home, keyed on every manifest file's - mtime (an in-place edit invalidates it); an unreadable manifest raises instead of silently - dropping the declaration.""" - from hermes_cli.config import platform_manifest_secret_envs, platform_manifest_stamp + profile B's children nor be missing from A's. Cached per home and keyed on every manifest's + file signature, so an edit, replacement or deletion takes effect on the next spawn. A plugin + manifest under ``plugins/platforms/`` that cannot be read raises. A partial scan (a plugin dir + or flat manifest unreadable or unparsable) keeps the names this home already had and is not + cached, so a failed discovery never releases a known denial and recovery is seen at once.""" + from hermes_cli.config import platform_manifest_secret_scan, platform_manifest_stamp from hermes_constants import get_hermes_home, hermes_home_key home = get_hermes_home() key, stamp = hermes_home_key(home), platform_manifest_stamp(home) cached = _HOME_ADAPTER_SECRET_CACHE.get(key) - if cached is None or cached[0] != stamp: - cached = (stamp, platform_manifest_secret_envs(home, strict=True) - _ADAPTER_SECRET_ENV) + if cached is None or cached[0] is None or cached[0] != stamp: + names, complete = platform_manifest_secret_scan(home) + names -= _ADAPTER_SECRET_ENV + if not complete and cached is not None: + names |= cached[1] + cached = (stamp if complete else None, names) _HOME_ADAPTER_SECRET_CACHE[key] = cached return cached[1]