fix(child-env): a partial plugin scan keeps the denials it already knew

andrexibiza (review on 60bdd5fbe3): skipping an unreadable plugin dir,
or an unreadable or unparsable flat manifest, replaced the home's
cached secret set with the partial result. A secret declared a moment
ago could then reach children while its manifest was unreadable. And
because the partial result was cached under the same file signature,
recovery could keep hitting it.

platform_manifest_secret_scan() now reports whether the scan was
complete. A partial scan unions in the names this home already had,
and is cached with no stamp, so the next spawn rescans and a recovery
is seen at once. A deleted plugin still releases its names, because
that scan is complete. An unreadable plugins/platforms/ manifest still
raises.
This commit is contained in:
kshitijk4poor
2026-09-28 18:20:48 +05:30
committed by kshitij
parent 9e26d34475
commit a0e20b60d9
3 changed files with 97 additions and 8 deletions

View File

@@ -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`` /

View File

@@ -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

View File

@@ -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]