fix(gateway): re-check every stacked skill against the platform-disabled list

_handle_message() re-checks a slash-skill command's per-platform disabled
status before dispatch, because get_skill_commands() only applies the
global disabled list at scan time. That check only covered the leading
skill: split_stacked_skill_commands() resolves additional /skill tokens
that follow it (stacked invocations, up to 5 skills, #57987), and
build_stacked_skill_invocation_message() loads every one of them via
_load_skill_payload() with no disabled-status check of any kind.

A message on a platform with skills.platform_disabled configured for a
given skill could still get that skill's full SKILL.md content injected
into the agent's context for the turn, as long as it was typed after an
allowed skill: `/allowed-skill /disabled-skill do X`.

Fix: after computing the stacked extra_keys, look up each one's skill
name and re-check it against the same get_disabled_skill_names(platform=)
set already used for the leading skill. If any stacked skill is disabled
for the platform, reject the whole invocation with the same style of
message the leading-skill check already returns, instead of partially
loading it.
This commit is contained in:
srojk34 2026-07-05 16:55:45 +03:00 committed by Teknium
parent 9d2ff58f5f
commit 04d732dc56
2 changed files with 184 additions and 0 deletions

View file

@ -9894,6 +9894,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception:
_build_stacked = None
extra_keys, stacked_instruction = [], user_instruction
if extra_keys and _plat:
# split_stacked_skill_commands() only resolves that
# each extra token is a KNOWN skill command — like
# get_skill_commands() itself, it has no per-platform
# view. Re-check every stacked skill (not just the
# leading one above) against the same disabled list,
# or a skill an operator disabled for this platform
# still gets its full content loaded via the stack.
from agent.skill_utils import get_disabled_skill_names as _get_plat_disabled
_plat_disabled = _get_plat_disabled(platform=_plat)
_disabled_extra = [
skill_cmds.get(k, {}).get("name", "")
for k in extra_keys
if skill_cmds.get(k, {}).get("name", "") in _plat_disabled
]
if _disabled_extra:
return (
f"The **{', '.join(_disabled_extra)}** skill(s) in this "
f"stacked invocation are disabled for {_plat}.\n"
f"Enable them with: `hermes skills config`"
)
if extra_keys and _build_stacked is not None:
stacked_result = _build_stacked(
[cmd_key, *extra_keys],

View file

@ -0,0 +1,163 @@
"""Regression test for stacked slash-skill invocations bypassing the
per-platform ``skills.platform_disabled`` gate.
``/skill-a /skill-b do XYZ`` loads every leading skill (up to 5), not just
the first (``agent.skill_commands.split_stacked_skill_commands`` /
``build_stacked_skill_invocation_message``). ``gateway.run.GatewayRunner.
_handle_message`` already re-checks the FIRST skill against the
per-platform disabled list before dispatch (``get_skill_commands()`` only
applies the *global* disabled list at scan time), but did not extend that
same check to the additional stacked skills a skill an operator disabled
for a given platform still had its full SKILL.md content injected into the
agent's context for that turn if it was stacked behind an allowed one.
"""
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import MessageEvent
from gateway.session import SessionEntry, SessionSource, build_session_key
def _make_source() -> SessionSource:
return SessionSource(
platform=Platform.TELEGRAM,
user_id="u1",
chat_id="c1",
user_name="tester",
chat_type="dm",
)
def _make_event(text: str) -> MessageEvent:
return MessageEvent(text=text, source=_make_source(), message_id="m1")
def _make_runner():
from gateway.run import GatewayRunner
runner = object.__new__(GatewayRunner)
runner.config = GatewayConfig(
platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
)
adapter = MagicMock()
adapter.send = AsyncMock()
runner.adapters = {Platform.TELEGRAM: adapter}
runner._voice_mode = {}
runner.hooks = SimpleNamespace(
emit=AsyncMock(),
emit_collect=AsyncMock(return_value=[]),
loaded_hooks=False,
)
session_entry = SessionEntry(
session_key=build_session_key(_make_source()),
session_id="sess-1",
created_at=datetime.now(),
updated_at=datetime.now(),
platform=Platform.TELEGRAM,
chat_type="dm",
)
runner.session_store = MagicMock()
runner.session_store.get_or_create_session.return_value = session_entry
runner.session_store.load_transcript.return_value = []
runner.session_store.has_any_sessions.return_value = True
runner.session_store.append_to_transcript = MagicMock()
runner.session_store.rewrite_transcript = MagicMock()
runner.session_store.update_session = MagicMock()
runner._running_agents = {}
runner._pending_messages = {}
runner._pending_approvals = {}
runner._session_db = None
runner._reasoning_config = None
runner._provider_routing = {}
runner._fallback_model = None
runner._show_reasoning = False
runner._is_user_authorized = lambda _source: True
runner._set_session_env = lambda _context: None
runner._should_send_voice_reply = lambda *_args, **_kwargs: False
from gateway.run import GatewayRunner as _GR
runner._session_key_for_source = _GR._session_key_for_source.__get__(runner, _GR)
return runner
def _make_skill(skills_dir, name, body="content"):
sd = skills_dir / name
sd.mkdir(parents=True, exist_ok=True)
(sd / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: desc {name}\n---\n\n# {name}\n\n{body}\n"
)
@pytest.fixture
def skills_env(tmp_path, monkeypatch):
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
import tools.skills_tool as skills_tool_module
monkeypatch.setattr(skills_tool_module, "SKILLS_DIR", skills_dir)
import agent.skill_commands as skill_commands_mod
skill_commands_mod._skill_commands = {}
skill_commands_mod._skill_commands_platform = None
return skills_dir
@pytest.mark.asyncio
async def test_stacked_second_skill_disabled_for_platform_is_blocked(monkeypatch, skills_env):
"""The whole stacked invocation is rejected when a NON-leading stacked
skill is disabled for the message's platform — it must not silently load
that skill's content just because only the first skill was checked."""
import gateway.run as gateway_run
import agent.skill_utils as skill_utils_mod
_make_skill(skills_env, "allowed-skill")
_make_skill(skills_env, "disabled-skill")
monkeypatch.setattr(
skill_utils_mod,
"get_disabled_skill_names",
lambda platform=None: {"disabled-skill"} if platform == "telegram" else set(),
)
monkeypatch.setattr(
gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
)
runner = _make_runner()
result = await runner._handle_message(
_make_event("/allowed-skill /disabled-skill do something")
)
assert result is not None
assert "disabled-skill" in result
assert "disabled for telegram" in result
@pytest.mark.asyncio
async def test_stacked_all_enabled_skills_still_load(monkeypatch, skills_env):
"""Positive control: the new platform-disabled check must not over-block
a stacked invocation where every skill is actually enabled."""
import gateway.run as gateway_run
import agent.skill_utils as skill_utils_mod
_make_skill(skills_env, "alpha-skill", body="ALPHA BODY MARKER")
_make_skill(skills_env, "beta-skill", body="BETA BODY MARKER")
monkeypatch.setattr(
skill_utils_mod, "get_disabled_skill_names", lambda platform=None: set()
)
monkeypatch.setattr(
gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
)
runner = _make_runner()
event = _make_event("/alpha-skill /beta-skill do something")
result = await runner._handle_message(event)
# Not rejected: the handler falls through to normal message processing
# with event.text rewritten to the combined stacked-skill payload.
assert result is None or "disabled for" not in result
assert "ALPHA BODY MARKER" in event.text
assert "BETA BODY MARKER" in event.text