feat(plugins): capability-gated ctx.platform_actions facade (#64176)

Minimal v1 platform action surface for plugins, routed through the live
gateway adapter registry — the sanctioned alternative to monkeypatching an
adapter:

- ctx.platform_actions.add_reaction(platform, chat_id, message_id, emoji)
- ctx.platform_actions.set_thread_title(platform, chat_id, thread_id, title)

Gated behind a new 'gateway.platform_actions' capability in
CAPABILITY_REGISTRY (legacy key plugins.entries.<id>.allow_platform_actions,
default OFF), re-checked on every call via plugin_capability_granted (the
#84912 consent registry). Verbs validate the adapter exists and is connected,
return structured {ok, error, detail} results with stable error codes, and
never raise into hook dispatch. Every action is audit-logged with plugin id,
verb, platform, and outcome.

Telegram routes to _set_reaction / rename_dm_topic; Discord to
fetch_message().add_reaction / rename_thread. No adapter handles or raw SDK
objects are exposed.

Docs: plugins.md platform-actions section with the security note and the
explicit raw-SDK-not-shipped statement.
This commit is contained in:
Teknium 2026-08-12 19:40:13 -07:00
parent 3b7c940208
commit e3983f91eb
4 changed files with 574 additions and 0 deletions

View File

@ -0,0 +1,238 @@
"""Capability-gated platform action facade for plugins (#64176, action half).
``ctx.platform_actions`` gives a plugin a *minimal*, versioned verb set for
acting on connected chat platforms through the live gateway adapter registry
no adapter handles, bot clients, or raw SDK objects are ever exposed.
Gating (fail closed, default OFF)
---------------------------------
Every verb checks ``plugin_capability_granted(plugin_id,
"gateway.platform_actions")`` at call time. The capability maps to the
``plugins.entries.<id>.allow_platform_actions`` legacy key and the #64228
consent registry (``granted_capabilities``). No grant structured
``capability_not_granted`` error, never an exception.
v1 verb set
-----------
* ``add_reaction(platform, chat_id, message_id, emoji)``
* ``set_thread_title(platform, chat_id, thread_id, title)``
Both return a structured result dict ``{"ok": True, ...}`` on success,
``{"ok": False, "error": <code>, "detail": <str>}`` on failure and never
raise into hook dispatch. Error codes are part of the v1 contract:
``capability_not_granted``, ``invalid_argument``, ``gateway_unavailable``,
``unknown_platform``, ``adapter_not_registered``, ``adapter_disconnected``,
``unsupported_platform_action``, ``action_failed``.
Raw SDK payload/handle access is deliberately NOT part of this surface; per
the #64176 round-2 correction it requires its own capability
(``gateway.raw_events``, #64228) and design.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
ACTIONS_CONTRACT_VERSION = 1
CAPABILITY_ID = "gateway.platform_actions"
def _err(code: str, detail: str = "") -> Dict[str, Any]:
result: Dict[str, Any] = {"ok": False, "error": code}
if detail:
result["detail"] = detail
return result
def _ok(**fields: Any) -> Dict[str, Any]:
result: Dict[str, Any] = {"ok": True}
result.update(fields)
return result
class PlatformActions:
"""Per-plugin facade over the live gateway adapter registry.
Instances are cheap and hold only the owning plugin id; the gateway
runner and adapters are resolved at call time so a facade created
before the gateway starts (plugin ``register()`` runs first) still
works once adapters connect.
"""
def __init__(self, plugin_id: str):
self._plugin_id = plugin_id
# -- shared plumbing ----------------------------------------------------
def _capability_granted(self) -> bool:
try:
from hermes_cli.plugin_capabilities import plugin_capability_granted
return plugin_capability_granted(self._plugin_id, CAPABILITY_ID)
except Exception:
# Ground rule: failure to read consent state = not granted.
logger.debug(
"platform_actions capability check failed for %s",
self._plugin_id, exc_info=True,
)
return False
def _resolve_adapter(self, platform: str):
"""Return ``(adapter, error_dict)``; exactly one is non-None."""
try:
from gateway.run import _gateway_runner_ref
runner = _gateway_runner_ref()
except Exception:
runner = None
if runner is None:
return None, _err(
"gateway_unavailable", "no gateway runner is active in this process"
)
try:
from gateway.config import Platform
platform_enum = Platform(str(platform).strip().lower())
except Exception:
return None, _err("unknown_platform", f"unknown platform {platform!r}")
adapter = getattr(runner, "adapters", {}).get(platform_enum)
if adapter is None:
return None, _err(
"adapter_not_registered",
f"no {platform_enum.value} adapter is registered",
)
try:
connected = bool(adapter.is_connected)
except Exception:
connected = False
if not connected:
return None, _err(
"adapter_disconnected",
f"the {platform_enum.value} adapter is not connected",
)
return adapter, None
def _gate(self, platform: str, **required: Any):
"""Run the shared gate chain. Returns ``(adapter, error_dict)``."""
if not self._capability_granted():
return None, _err(
"capability_not_granted",
f"plugin {self._plugin_id!r} lacks the {CAPABILITY_ID!r} "
"capability (grant via consent flow or "
f"plugins.entries.{self._plugin_id}.allow_platform_actions)",
)
for name, value in required.items():
if not isinstance(value, str) or not value.strip():
return None, _err(
"invalid_argument", f"{name} must be a non-empty string"
)
return self._resolve_adapter(platform)
# -- v1 verbs -----------------------------------------------------------
async def add_reaction(
self, platform: str, chat_id: str, message_id: str, emoji: str
) -> Dict[str, Any]:
"""Add/set an emoji reaction on a platform message.
Telegram note: the Bot API *sets* the bot's reaction (replacing a
previous one) rather than stacking, per ``set_message_reaction``.
"""
adapter, error = self._gate(
platform, chat_id=chat_id, message_id=message_id, emoji=emoji
)
if error is not None or adapter is None:
self._audit("add_reaction", platform, error or _err("gateway_unavailable"))
return error or _err("gateway_unavailable")
try:
if getattr(adapter.platform, "value", None) == "telegram":
done = await adapter._set_reaction(chat_id, message_id, emoji)
result = (
_ok(action="add_reaction")
if done
else _err("action_failed", "telegram set_message_reaction failed")
)
elif getattr(adapter.platform, "value", None) == "discord":
result = await self._discord_add_reaction(
adapter, chat_id, message_id, emoji
)
else:
result = _err(
"unsupported_platform_action",
f"add_reaction is not implemented for {platform}",
)
except Exception as exc:
result = _err("action_failed", str(exc)[:512])
self._audit("add_reaction", platform, result)
return result
async def set_thread_title(
self, platform: str, chat_id: str, thread_id: str, title: str
) -> Dict[str, Any]:
"""Rename a thread / forum topic.
Discord ignores ``chat_id`` (thread ids are globally addressable);
Telegram requires it (``edit_forum_topic`` is chat-scoped).
"""
adapter, error = self._gate(
platform, chat_id=chat_id, thread_id=thread_id, title=title
)
if error is not None or adapter is None:
self._audit("set_thread_title", platform, error or _err("gateway_unavailable"))
return error or _err("gateway_unavailable")
try:
if getattr(adapter.platform, "value", None) == "telegram":
await adapter.rename_dm_topic(chat_id, int(thread_id), title)
result = _ok(action="set_thread_title")
elif getattr(adapter.platform, "value", None) == "discord":
done = await adapter.rename_thread(thread_id, title)
result = (
_ok(action="set_thread_title")
if done
else _err("action_failed", "discord thread rename failed")
)
else:
result = _err(
"unsupported_platform_action",
f"set_thread_title is not implemented for {platform}",
)
except Exception as exc:
result = _err("action_failed", str(exc)[:512])
self._audit("set_thread_title", platform, result)
return result
# -- per-platform helpers -------------------------------------------------
@staticmethod
async def _discord_add_reaction(
adapter: Any, chat_id: str, message_id: str, emoji: str
) -> Dict[str, Any]:
client = getattr(adapter, "_client", None)
if client is None:
return _err("adapter_disconnected", "discord client unavailable")
try:
channel_id = int(str(chat_id))
msg_id = int(str(message_id))
except (TypeError, ValueError):
return _err("invalid_argument", "discord ids must be numeric")
channel = client.get_channel(channel_id)
if channel is None:
channel = await client.fetch_channel(channel_id)
message = await channel.fetch_message(msg_id)
await message.add_reaction(emoji)
return _ok(action="add_reaction")
def _audit(self, verb: str, platform: str, result: Dict[str, Any]) -> None:
"""Every platform action is logged (the #64176 'all actions logged' rule)."""
logger.info(
"platform_action plugin=%s verb=%s platform=%s ok=%s%s",
self._plugin_id,
verb,
platform,
result.get("ok"),
"" if result.get("ok") else f" error={result.get('error')}",
)

View File

@ -25,6 +25,7 @@ Capability id Legacy config gate (``plugins.entries.<id>.…``)
``llm.agent_id_override`` ``llm.allow_agent_id_override``
``llm.profile_override`` ``llm.allow_profile_override``
``llm.task_override`` ``llm.allow_task_override``
``gateway.platform_actions`` ``allow_platform_actions``
=========================== ==================================================
The legacy ``allow_*`` keys keep working verbatim (deprecated but honored):
@ -119,6 +120,14 @@ CAPABILITY_REGISTRY: Dict[str, CapabilitySpec] = {
"task lanes"
),
),
CapabilitySpec(
id="gateway.platform_actions",
legacy_path=("allow_platform_actions",),
description=(
"Act on connected chat platforms as the gateway bot "
"(add reactions, rename threads) via ctx.platform_actions"
),
),
)
}

View File

@ -0,0 +1,279 @@
"""Tests for the capability-gated platform action facade (#64176, action half).
Covers:
* gate default-off: no grant ``capability_not_granted`` structured error,
no adapter touched
* capability grant honored via ``granted_capabilities`` AND via the legacy
``allow_platform_actions`` config key
* unknown platform / unregistered adapter / disconnected adapter structured
errors, never exceptions
* verbs route to the right adapter primitives (telegram ``_set_reaction`` /
``rename_dm_topic``; discord ``rename_thread``)
* adapter-layer exceptions surface as ``action_failed`` results, never raise
* ``ctx.platform_actions`` facade is bound to the plugin id
"""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.config import Platform
from hermes_cli.platform_actions import CAPABILITY_ID, PlatformActions
from hermes_cli.plugin_capabilities import CAPABILITY_REGISTRY
def _grant(granted: bool):
"""Patch the capability check the facade performs."""
return patch(
"hermes_cli.plugin_capabilities.plugin_capability_granted",
return_value=granted,
)
def _runner_with(adapters: dict):
runner = SimpleNamespace(adapters=adapters)
return patch("gateway.run._gateway_runner_ref", lambda: runner)
def _telegram_adapter(connected=True):
a = MagicMock()
a.platform = Platform.TELEGRAM
a.is_connected = connected
a._set_reaction = AsyncMock(return_value=True)
a.rename_dm_topic = AsyncMock(return_value=None)
return a
def _discord_adapter(connected=True):
a = MagicMock()
a.platform = Platform.DISCORD
a.is_connected = connected
a.rename_thread = AsyncMock(return_value=True)
return a
class TestCapabilityRegistry:
def test_gateway_platform_actions_registered(self):
spec = CAPABILITY_REGISTRY.get("gateway.platform_actions")
assert spec is not None
assert spec.legacy_path == ("allow_platform_actions",)
assert spec.description
def test_facade_uses_registered_id(self):
assert CAPABILITY_ID == "gateway.platform_actions"
class TestGateDefaultOff:
def test_no_grant_returns_structured_error(self):
actions = PlatformActions("some-plugin")
adapter = _telegram_adapter()
with _grant(False), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(
actions.add_reaction("telegram", "123", "456", "\U0001F44D")
)
assert result["ok"] is False
assert result["error"] == "capability_not_granted"
adapter._set_reaction.assert_not_awaited()
def test_default_config_is_off_via_real_capability_check(self):
"""No patching of the check itself: an empty config entry denies."""
actions = PlatformActions("some-plugin")
with patch(
"hermes_cli.plugin_capabilities._plugin_entry", return_value={}
):
result = asyncio.run(
actions.set_thread_title("telegram", "1", "2", "t")
)
assert result == {
"ok": False,
"error": "capability_not_granted",
"detail": result["detail"],
}
def test_legacy_allow_platform_actions_key_grants(self):
actions = PlatformActions("some-plugin")
adapter = _telegram_adapter()
with patch(
"hermes_cli.plugin_capabilities._plugin_entry",
return_value={"allow_platform_actions": True},
), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(
actions.add_reaction("telegram", "123", "456", "\U0001F44D")
)
assert result["ok"] is True
def test_granted_capabilities_list_grants(self):
actions = PlatformActions("some-plugin")
adapter = _telegram_adapter()
with patch(
"hermes_cli.plugin_capabilities._plugin_entry",
return_value={"granted_capabilities": ["gateway.platform_actions"]},
), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(
actions.add_reaction("telegram", "123", "456", "\U0001F44D")
)
assert result["ok"] is True
def test_capability_check_failure_fails_closed(self):
actions = PlatformActions("some-plugin")
with patch(
"hermes_cli.plugin_capabilities.plugin_capability_granted",
side_effect=RuntimeError("corrupt config"),
):
result = asyncio.run(
actions.add_reaction("telegram", "1", "2", "x")
)
assert result["error"] == "capability_not_granted"
class TestStructuredErrors:
def test_no_gateway_runner(self):
actions = PlatformActions("p")
with _grant(True), patch("gateway.run._gateway_runner_ref", lambda: None):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", "x"))
assert result["error"] == "gateway_unavailable"
def test_unknown_platform(self):
actions = PlatformActions("p")
with _grant(True), _runner_with({}):
result = asyncio.run(actions.add_reaction("smoke-signals", "1", "2", "x"))
assert result["error"] == "unknown_platform"
def test_adapter_not_registered(self):
actions = PlatformActions("p")
with _grant(True), _runner_with({}):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", "x"))
assert result["error"] == "adapter_not_registered"
def test_adapter_disconnected(self):
actions = PlatformActions("p")
adapter = _telegram_adapter(connected=False)
with _grant(True), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", "x"))
assert result["error"] == "adapter_disconnected"
adapter._set_reaction.assert_not_awaited()
@pytest.mark.parametrize("bad", ["", " ", None, 123])
def test_invalid_arguments(self, bad):
actions = PlatformActions("p")
with _grant(True):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", bad))
assert result["error"] == "invalid_argument"
def test_adapter_exception_becomes_action_failed(self):
actions = PlatformActions("p")
adapter = _telegram_adapter()
adapter._set_reaction = AsyncMock(side_effect=RuntimeError("api down"))
with _grant(True), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", "x"))
assert result["ok"] is False
assert result["error"] == "action_failed"
assert "api down" in result["detail"]
def test_unsupported_platform_action(self):
actions = PlatformActions("p")
adapter = MagicMock()
adapter.platform = Platform.SLACK
adapter.is_connected = True
with _grant(True), _runner_with({Platform.SLACK: adapter}):
result = asyncio.run(actions.add_reaction("slack", "1", "2", "x"))
assert result["error"] == "unsupported_platform_action"
class TestVerbRouting:
def test_telegram_add_reaction_routes_to_set_reaction(self):
actions = PlatformActions("p")
adapter = _telegram_adapter()
with _grant(True), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(
actions.add_reaction("telegram", "-100123", "456", "\U0001F44D")
)
assert result == {"ok": True, "action": "add_reaction"}
adapter._set_reaction.assert_awaited_once_with("-100123", "456", "\U0001F44D")
def test_telegram_set_reaction_false_is_action_failed(self):
actions = PlatformActions("p")
adapter = _telegram_adapter()
adapter._set_reaction = AsyncMock(return_value=False)
with _grant(True), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(actions.add_reaction("telegram", "1", "2", "x"))
assert result["error"] == "action_failed"
def test_telegram_set_thread_title_routes_to_rename_dm_topic(self):
actions = PlatformActions("p")
adapter = _telegram_adapter()
with _grant(True), _runner_with({Platform.TELEGRAM: adapter}):
result = asyncio.run(
actions.set_thread_title("telegram", "123", "42", "New title")
)
assert result == {"ok": True, "action": "set_thread_title"}
adapter.rename_dm_topic.assert_awaited_once_with("123", 42, "New title")
def test_discord_set_thread_title_routes_to_rename_thread(self):
actions = PlatformActions("p")
adapter = _discord_adapter()
with _grant(True), _runner_with({Platform.DISCORD: adapter}):
result = asyncio.run(
actions.set_thread_title("discord", "555", "321", "Renamed")
)
assert result == {"ok": True, "action": "set_thread_title"}
adapter.rename_thread.assert_awaited_once_with("321", "Renamed")
def test_discord_rename_false_is_action_failed(self):
actions = PlatformActions("p")
adapter = _discord_adapter()
adapter.rename_thread = AsyncMock(return_value=False)
with _grant(True), _runner_with({Platform.DISCORD: adapter}):
result = asyncio.run(
actions.set_thread_title("discord", "555", "321", "Renamed")
)
assert result["error"] == "action_failed"
def test_discord_add_reaction_fetches_and_reacts(self):
actions = PlatformActions("p")
adapter = _discord_adapter()
message = MagicMock()
message.add_reaction = AsyncMock()
channel = MagicMock()
channel.fetch_message = AsyncMock(return_value=message)
client = MagicMock()
client.get_channel = MagicMock(return_value=channel)
adapter._client = client
with _grant(True), _runner_with({Platform.DISCORD: adapter}):
result = asyncio.run(
actions.add_reaction("discord", "555", "456", "\U0001F44D")
)
assert result == {"ok": True, "action": "add_reaction"}
channel.fetch_message.assert_awaited_once_with(456)
message.add_reaction.assert_awaited_once_with("\U0001F44D")
def test_discord_add_reaction_non_numeric_ids(self):
actions = PlatformActions("p")
adapter = _discord_adapter()
adapter._client = MagicMock()
with _grant(True), _runner_with({Platform.DISCORD: adapter}):
result = asyncio.run(
actions.add_reaction("discord", "not-a-number", "456", "x")
)
assert result["error"] == "invalid_argument"
class TestPluginContextWiring:
def test_ctx_platform_actions_bound_to_plugin_id(self):
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
manager = PluginManager()
ctx = PluginContext(
PluginManifest(name="actions-fixture", source="user"), manager,
)
facade = ctx.platform_actions
assert isinstance(facade, PlatformActions)
assert facade._plugin_id == "actions-fixture"
# Cached: property returns the same instance.
assert ctx.platform_actions is facade

View File

@ -391,6 +391,7 @@ working but are **deprecated** in favor of the consent flow:
| `llm.agent_id_override` | `llm.allow_agent_id_override` |
| `llm.profile_override` | `llm.allow_profile_override` |
| `llm.task_override` | `llm.allow_task_override` |
| `gateway.platform_actions` | `allow_platform_actions` |
A gate is open when *either* the capability is granted *or* the legacy key is
set — existing configs keep working unchanged.
@ -403,6 +404,53 @@ not a code audit, and Hermes has not reviewed the plugin's code. Only install
plugins from sources you trust.
:::
### Platform actions
`ctx.platform_actions` gives a plugin a minimal, capability-gated verb set for
acting on connected chat platforms through the live gateway adapter registry —
the sanctioned alternative to monkeypatching an adapter. **It is off by
default**: every call re-checks the `gateway.platform_actions` capability
(legacy key `plugins.entries.<id>.allow_platform_actions`), and an ungranted
call returns a structured error instead of acting.
v1 verbs (both `async`, both return a plain dict, and neither ever raises into
hook dispatch):
```python
result = await ctx.platform_actions.add_reaction(
platform="telegram", chat_id="-100123", message_id="456", emoji="👍",
)
result = await ctx.platform_actions.set_thread_title(
platform="discord", chat_id="123", thread_id="456", title="New title",
)
if not result["ok"]:
print(result["error"], result.get("detail"))
```
Success is `{"ok": True, "action": <verb>}`. Failures are
`{"ok": False, "error": <code>, "detail": <str>}` with stable error codes:
`capability_not_granted`, `invalid_argument`, `gateway_unavailable`,
`unknown_platform`, `adapter_not_registered`, `adapter_disconnected`,
`unsupported_platform_action`, `action_failed`. Actions validate that the
target adapter exists and is connected before acting; a disconnected or
missing adapter degrades to a structured error, never an exception.
Platforms supported in v1: Telegram and Discord. Telegram's `add_reaction`
*sets* the bot's reaction (the Bot API replaces a previous bot reaction rather
than stacking). Every action — allowed or denied — is written to the log with
the plugin id, verb, platform, and outcome.
:::warning Security note
Platform actions are a **messaging-as-the-bot power**: a granted plugin can
react and rename threads in any chat the gateway bot can reach, not just the
chat that triggered the hook. Grant `gateway.platform_actions` only to plugins
you trust, and prefer plugins that document exactly which actions they take.
Raw platform SDK payload/handle access is deliberately **not** part of this
surface — per the #64176 round-2 design correction it requires its own
capability (`gateway.raw_events`) with a "no stability guarantee" label and a
separate design, and has not shipped.
:::
### Discovering community plugins
`hermes plugins search <term>` searches the **community plugin index** — a