fix: a user plugin cannot mark a core platform trusted_inbound

PluginContext.register_platform is last-writer-wins, so a user plugin could re-register 'telegram'
with trusted_inbound=True and every sender would skip the allowlist and pairing. The flag is now
refused for a platform core ships (a static Platform member or a bundled plugins/platforms/ plugin)
unless the registering plugin is bundled. Home Assistant is no longer either, so its catalog plugin
keeps it. `hermes plugins validate` prints a line for every trusted_inbound platform (and fails one
naming a core platform).

Also: the MCP collision test docstring no longer refers to the toolset it used to collide with.
This commit is contained in:
teknium1
2026-10-03 15:19:11 -07:00
committed by Teknium
parent 3564e9cc24
commit c19dd99597
5 changed files with 49 additions and 5 deletions

View File

@@ -120,6 +120,16 @@ class PlatformEntry:
shared_env_prefixes: tuple = ()
def core_ships_platform(name: str) -> bool:
"""*name* is a platform core ships an adapter for: a static ``Platform`` member or a bundled
``plugins/platforms/`` plugin. A user plugin may replace such an adapter but never mark it
``trusted_inbound``: its senders are people, and that flag waives allowlists and pairing."""
from gateway.config import Platform
value = str(name or "").strip().lower()
return value in {Platform[member].value for member in Platform._member_names_} or (
value in Platform._scan_bundled_plugin_platforms()[0])
class PlatformRegistry:
"""Central registry of platform adapters. Registrations are serialized; concurrent
lazy lookups share an in-flight event while the loader runs outside the registry lock."""

View File

@@ -206,7 +206,7 @@ options = json.loads(sys.argv[3])
context_methods = set(options["context_methods"])
provider_kind = options["kind"] == "model-provider"
recorded = {"tools": [], "hooks": [], "middleware": [], "commands": [], "providers": []}
recorded = {"tools": [], "hooks": [], "middleware": [], "commands": [], "providers": [], "trusted_inbound": []}
class RecordingContext:
@@ -229,6 +229,10 @@ class RecordingContext:
def register_cli_command(self, name, *args, **kwargs):
recorded["commands"].append(str(name))
def register_platform(self, name, *args, **kwargs):
if kwargs.get("trusted_inbound"):
recorded["trusted_inbound"].append(str(name))
def get_config(self, key, default=None):
# Mirrors PluginContext.get_config with no config on disk: the DEFAULT, never None —
# plugins do `int(ctx.get_config("timeout", 180))` in register().
@@ -518,6 +522,7 @@ def validate_plugin_dir(plugin_dir: Path) -> ValidationReport:
_check_python_dependencies(report, plugin_dir)
recorded = _check_capabilities(report, manifest, plugin_dir)
_check_builtin_collisions(report, manifest, recorded)
_check_trusted_inbound(report, recorded)
_check_security_scan(report, plugin_dir)
check_core_override(report, plugin_dir)
check_desktop_surface(report, plugin_dir)
@@ -527,6 +532,17 @@ def validate_plugin_dir(plugin_dir: Path) -> ValidationReport:
return report
def _check_trusted_inbound(report: ValidationReport, recorded: Optional[dict]) -> None:
"""Surface ``trusted_inbound`` platforms (their events skip user allowlists and pairing); one
naming a core platform fails, as the loader refuses it."""
from gateway.platform_registry import core_ships_platform
for name in (recorded or {}).get("trusted_inbound") or []:
if core_ships_platform(name):
report.add("trusted inbound", False, f"platform '{name}' ships with core; trusted_inbound is refused for it")
else:
report.add("trusted inbound", True, f"platform '{name}' is trusted_inbound: its events skip user allowlists and pairing")
_LOADABLE_ENTRYPOINTS = ("__init__.py", "desktop/plugin.js", "plugin.json")

View File

@@ -815,7 +815,9 @@ class PluginContext:
it freely); an ACTIVE installer goes in ``ensure_deps_fn`` (called from ``create_adapter()`` when
``check_fn`` is False). Extra kwargs (``setup_fn``, ``emoji``, ``allowed_users_env``,
``platform_hint``, ``ensure_deps_fn``) forward to ``PlatformEntry``; unknown keys raise TypeError."""
from gateway.platform_registry import platform_registry, PlatformEntry
from gateway.platform_registry import core_ships_platform, platform_registry, PlatformEntry
if entry_kwargs.get("trusted_inbound") and self.manifest.source != "bundled" and core_ships_platform(name):
raise self._refuse(f"core platform '{name}' with trusted_inbound (it would waive allowlists and pairing)")
entry_kwargs.setdefault("plugin_name", self.manifest.name)
entry = PlatformEntry(
name=name, label=label, adapter_factory=adapter_factory, check_fn=check_fn,

View File

@@ -69,3 +69,19 @@ def test_display_tier_supplies_builtin_defaults_below_user_overrides(plugin_plat
def test_platform_entry_seam_defaults_are_inert():
entry = PlatformEntry(name="x", label="X", adapter_factory=lambda cfg: None, check_fn=lambda: True)
assert (entry.trusted_inbound, entry.display_tier, entry.shared_env_prefixes) == (False, "", ())
def test_user_plugin_cannot_mark_a_core_platform_trusted_inbound():
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
ctx = PluginContext(PluginManifest(name="seamevil", key="seamevil", source="user"),
PluginManager(scope_key=platform_registry.current_scope_key()))
with pytest.raises(ValueError, match="trusted_inbound"):
ctx.register_platform("telegram", "Telegram", adapter_factory=lambda cfg: None, check_fn=lambda: True,
trusted_inbound=True)
# Control: a platform core does not ship (Home Assistant's case) may declare it.
try:
ctx.register_platform("seameventbus", "Bus", adapter_factory=lambda cfg: None, check_fn=lambda: True,
trusted_inbound=True)
assert platform_registry.get("seameventbus").trusted_inbound is True
finally:
platform_registry.unregister("seameventbus", scope=platform_registry.current_scope_key())

View File

@@ -39,9 +39,9 @@ class TestRegisterServerTools:
"""An MCP server named after a built-in toolset must not be shadowed.
Regression: an MCP server registered under a built-in toolset's name
(originally `browser`; `browser` here) had its tools silently
dropped because get_toolset() returned the static definition without
consulting the alias registered by _register_server_tools().
(`browser` here) had its tools silently dropped because get_toolset()
returned the static definition without consulting the alias
registered by _register_server_tools().
"""
from toolsets import TOOLSETS, get_toolset, resolve_toolset