fix(gateway): /help and /commands show a gated non-admin only the commands they can run
Both catalog handlers called the shared executor without the caller's slash-access policy, so a non-admin under allow_admin_from saw every admin-only command and then hit the refusal on each one. The handlers now pass the policy's runnable set (the /help+/whoami floor plus user_allowed_commands) as `allowed_commands`; gateway_help_lines filters on it and skill commands are hidden for gated users. Admins and ungated scopes are unchanged (option absent -> full catalog). Part of #117217
This commit is contained in:
@@ -585,14 +585,27 @@ class GatewaySlashCommandsMixin(
|
||||
"""Handle /version — show the running Hermes Agent version."""
|
||||
return _execute("version").text
|
||||
|
||||
def _catalog_options(self, event: MessageEvent) -> dict:
|
||||
"""``allowed_commands`` for /help and /commands when the caller is a gated non-admin:
|
||||
the slash-access floor + ``user_allowed_commands`` (mirrors /whoami), so the catalog
|
||||
never advertises commands ``_check_slash_access`` would refuse. Admins / ungated -> {}."""
|
||||
from gateway.slash_access import policy_for_source
|
||||
source = event.source
|
||||
policy = policy_for_source(self.config, source)
|
||||
if policy.enabled and not policy.is_admin(source.user_id if source else None):
|
||||
return {"allowed_commands": {"help", "whoami", *policy.user_allowed_commands}}
|
||||
return {}
|
||||
|
||||
async def _handle_help_command(self, event: MessageEvent) -> str:
|
||||
"""Handle /help command - list available commands."""
|
||||
return self._telegramized_command_reply(event, _execute("help").text)
|
||||
return self._telegramized_command_reply(
|
||||
event, _execute("help", options=self._catalog_options(event)).text)
|
||||
|
||||
async def _handle_commands_command(self, event: MessageEvent) -> str:
|
||||
# Page size is a surface parameter (Telegram messages are shorter).
|
||||
page_size = 15 if event.source.platform == Platform.TELEGRAM else 20
|
||||
reply = _execute("commands", args=event.get_command_args(), options={"page_size": page_size})
|
||||
options = {"page_size": page_size, **self._catalog_options(event)}
|
||||
reply = _execute("commands", args=event.get_command_args(), options=options)
|
||||
return self._telegramized_command_reply(event, reply.text)
|
||||
|
||||
async def _handle_set_home_command(self, event: MessageEvent) -> str:
|
||||
|
||||
@@ -9,7 +9,9 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from utils import is_truthy_value
|
||||
from hermes_constants import INDICATOR_STYLES
|
||||
@@ -462,13 +464,21 @@ def _is_gateway_available(cmd: CommandDef, config_overrides: set[str] | None = N
|
||||
return cmd.name in overrides
|
||||
|
||||
|
||||
def gateway_help_lines() -> list[str]:
|
||||
"""Generate gateway help text lines from the registry."""
|
||||
def gateway_help_lines(allowed: Optional[Iterable[str]] = None) -> list[str]:
|
||||
"""Generate gateway help text lines from the registry.
|
||||
|
||||
``allowed`` (canonical names) restricts the catalog to what the caller may run -- the
|
||||
gateway passes a non-admin's slash-access floor + ``user_allowed_commands`` so /help never
|
||||
advertises admin-only commands the dispatcher would then refuse.
|
||||
"""
|
||||
overrides = _resolve_config_gates()
|
||||
allowed_set = None if allowed is None else set(allowed)
|
||||
lines: list[str] = []
|
||||
for cmd in COMMAND_REGISTRY:
|
||||
if not _is_gateway_available(cmd, overrides):
|
||||
continue
|
||||
if allowed_set is not None and cmd.name not in allowed_set:
|
||||
continue
|
||||
args = f" {cmd.args_hint}" if cmd.args_hint else ""
|
||||
# Skip internal aliases like reload_mcp (underscore variant of the name).
|
||||
alias_parts = [f"`/{a}`" for a in cmd.aliases
|
||||
|
||||
@@ -105,8 +105,11 @@ def _exec_help(ctx: CommandContext) -> CommandReply:
|
||||
"""Core gateway /help body (pre platform mention decoration)."""
|
||||
from agent.i18n import t
|
||||
from hermes_cli.commands import gateway_help_lines
|
||||
lines = [t("gateway.help.header"), *gateway_help_lines()]
|
||||
skill_cmds = _skill_commands()
|
||||
# ``allowed_commands`` (gateway, non-admin caller): only the commands the slash-access
|
||||
# policy lets this user run; skill commands are hidden too since the gate refuses them.
|
||||
allowed = ctx.options.get("allowed_commands")
|
||||
lines = [t("gateway.help.header"), *gateway_help_lines(allowed)]
|
||||
skill_cmds = _skill_commands() if allowed is None else {}
|
||||
try:
|
||||
if skill_cmds:
|
||||
lines.append(t("gateway.help.skill_header", count=len(skill_cmds)))
|
||||
@@ -132,8 +135,9 @@ def _exec_commands(ctx: CommandContext) -> CommandReply:
|
||||
except ValueError:
|
||||
return CommandReply(t("gateway.commands.usage"), format="markdown")
|
||||
|
||||
entries = list(gateway_help_lines())
|
||||
skill_cmds = _skill_commands()
|
||||
allowed = ctx.options.get("allowed_commands")
|
||||
entries = list(gateway_help_lines(allowed))
|
||||
skill_cmds = _skill_commands() if allowed is None else {}
|
||||
try:
|
||||
if skill_cmds:
|
||||
entries.extend(["", t("gateway.commands.skill_header")])
|
||||
|
||||
@@ -133,6 +133,23 @@ async def test_whoami_non_admin_lists_runnable_commands():
|
||||
assert "/model" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_help_non_admin_lists_only_runnable_commands():
|
||||
"""/help for a gated non-admin renders the floor + user_allowed_commands, never the
|
||||
admin-only catalog the dispatcher would then refuse; admins keep the full list."""
|
||||
runner = _make_runner(
|
||||
platform_extra={
|
||||
"allow_admin_from": ["111"],
|
||||
"user_allowed_commands": ["status"],
|
||||
}
|
||||
)
|
||||
user = await runner._handle_message(_make_event("/help", _make_source(user_id="999")))
|
||||
assert "`/help" in user and "`/whoami" in user and "`/status" in user
|
||||
assert "`/model" not in user and "`/restart" not in user
|
||||
admin = await runner._handle_message(_make_event("/help", _make_source(user_id="111")))
|
||||
assert "`/model" in admin and "`/restart" in admin
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Gate denial — admin-only command attempted by non-admin
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -436,7 +436,7 @@ suppress unrelated notices.
|
||||
|
||||
#### Inspecting your access
|
||||
|
||||
Use `/whoami` from any platform to see the active scope, your tier (admin / user / unrestricted), and which slash commands you can run. See the [Telegram](./telegram.md#slash-command-access-control) and [Discord](./discord.md#slash-command-access-control) pages for platform-specific examples.
|
||||
Use `/whoami` from any platform to see the active scope, your tier (admin / user / unrestricted), and which slash commands you can run. When an admin list is configured, `/help` and `/commands` show a non-admin only the commands they can actually run (`/help`, `/whoami`, plus `user_allowed_commands`); admins see the full catalog. See the [Telegram](./telegram.md#slash-command-access-control) and [Discord](./discord.md#slash-command-access-control) pages for platform-specific examples.
|
||||
|
||||
## Redirecting the Agent
|
||||
|
||||
|
||||
Reference in New Issue
Block a user