fix: harden gateway slash command security (#127)

* fix: harden gateway slash command security

* fix: add secure remote admin opt-in for gateway commands

ohmo/cli.py

@@ -306,6 +306,22 @@ def _run_gateway_config_wizard(workspace: str | Path) -> GatewayConfig:
         "Send tool hints to channels?",
         default=existing.send_tool_hints,
     )
+    allow_remote_admin_commands = _confirm_prompt(
+        "Allow explicitly listed administrative slash commands from remote channels?",
+        default=existing.allow_remote_admin_commands,
+    )
+    default_allowlist = ", ".join(existing.allowed_remote_admin_commands)
+    allowed_remote_admin_commands: list[str] = []
+    if allow_remote_admin_commands:
+        allowlist_raw = _text_prompt(
+            "Allowed remote admin commands (comma-separated, e.g. permissions, plan)",
+            default=default_allowlist,
+        )
+        allowed_remote_admin_commands = [
+            item.strip().lstrip("/")
+            for item in allowlist_raw.split(",")
+            if item.strip()
+        ]
     config = existing.model_copy(
         update={
             "provider_profile": provider_profile,
@@ -313,6 +329,8 @@ def _run_gateway_config_wizard(workspace: str | Path) -> GatewayConfig:
             "channel_configs": channel_configs,
             "send_progress": send_progress,
             "send_tool_hints": send_tool_hints,
+            "allow_remote_admin_commands": allow_remote_admin_commands,
+            "allowed_remote_admin_commands": allowed_remote_admin_commands,
         }
     )
     save_gateway_config(config, workspace)
@@ -328,6 +346,13 @@ def _print_gateway_config_summary(config: GatewayConfig) -> None:
         )
     else:
         print(f"Configured provider_profile={config.provider_profile}; no channels enabled yet.")
+    if config.allow_remote_admin_commands and config.allowed_remote_admin_commands:
+        print(
+            "Remote admin opt-in enabled for: "
+            + ", ".join(f"/{name}" for name in config.allowed_remote_admin_commands)
+        )
+    else:
+        print("Remote admin commands remain local-only.")
 
 
 def _maybe_restart_gateway(*, cwd: str | Path, workspace: str | Path) -> None:

ohmo/gateway/models.py

@@ -15,6 +15,8 @@ class GatewayConfig(BaseModel):
     send_tool_hints: bool = True
     permission_mode: str = "default"
     sandbox_enabled: bool = False
+    allow_remote_admin_commands: bool = False
+    allowed_remote_admin_commands: list[str] = Field(default_factory=list)
     log_level: str = "INFO"
     channel_configs: dict[str, dict] = Field(default_factory=dict)
 

ohmo/gateway/runtime.py

@@ -12,7 +12,7 @@ import os
 import string
 
 from openharness.channels.bus.events import InboundMessage
-from openharness.commands import CommandContext
+from openharness.commands import CommandContext, CommandResult
 from openharness.engine.messages import ConversationMessage, ImageBlock, TextBlock
 from openharness.engine.query import MaxTurnsExceeded
 from openharness.engine.stream_events import (
@@ -27,6 +27,7 @@ from openharness.engine.stream_events import (
 from openharness.prompts import build_runtime_system_prompt
 from openharness.ui.runtime import RuntimeBundle, _last_user_text, build_runtime, close_runtime, start_runtime
 
+from ohmo.gateway.config import load_gateway_config
 from ohmo.prompts import build_ohmo_system_prompt
 from ohmo.session_storage import OhmoSessionBackend
 from ohmo.workspace import get_plugins_dir, get_skills_dir, initialize_workspace
@@ -81,6 +82,7 @@ class OhmoSessionRuntimePool:
         self._model = model
         self._max_turns = max_turns
         self._workspace = initialize_workspace(workspace)
+        self._gateway_config = load_gateway_config(self._workspace)
         self._session_backend = OhmoSessionBackend(self._workspace)
         self._bundles: dict[str, RuntimeBundle] = {}
 
@@ -88,6 +90,18 @@ class OhmoSessionRuntimePool:
     def active_sessions(self) -> int:
         return len(self._bundles)
 
+    def _remote_admin_allowed(self, command) -> bool:
+        if not getattr(command, "remote_admin_opt_in", False):
+            return False
+        if not self._gateway_config.allow_remote_admin_commands:
+            return False
+        allowed = {
+            str(name).strip().lower()
+            for name in self._gateway_config.allowed_remote_admin_commands
+            if str(name).strip()
+        }
+        return command.name.lower() in allowed
+
     async def get_bundle(self, session_key: str, latest_user_prompt: str | None = None) -> RuntimeBundle:
         """Return an existing bundle or create a new one."""
         bundle = self._bundles.get(session_key)
@@ -149,6 +163,29 @@ class OhmoSessionRuntimePool:
         parsed = bundle.commands.lookup(user_prompt)
         if parsed is not None and not message.media:
             command, args = parsed
+            remote_allowed = getattr(command, "remote_invocable", True)
+            if not remote_allowed and self._remote_admin_allowed(command):
+                remote_allowed = True
+                logger.warning(
+                    "ohmo gateway remote administrative command accepted channel=%s chat_id=%s sender_id=%s command=%s",
+                    message.channel,
+                    message.chat_id,
+                    message.sender_id,
+                    command.name,
+                )
+            if not remote_allowed:
+                result = CommandResult(
+                    message=f"/{command.name} is only available in the local OpenHarness UI."
+                )
+                async for update in self._stream_command_result(
+                    bundle=bundle,
+                    message=message,
+                    session_key=session_key,
+                    user_prompt=user_prompt,
+                    result=result,
+                ):
+                    yield update
+                return
             result = await command.handler(
                 args,
                 CommandContext(

ohmo/gateway/service.py

@@ -43,6 +43,11 @@ class OhmoGatewayService:
         root = initialize_workspace(self._workspace)
         os.environ["OHMO_WORKSPACE"] = str(root)
         self._config = load_gateway_config(self._workspace)
+        if self._config.allow_remote_admin_commands and self._config.allowed_remote_admin_commands:
+            logger.warning(
+                "ohmo gateway remote administrative commands enabled commands=%s",
+                ",".join(self._config.allowed_remote_admin_commands),
+            )
         self._bus = MessageBus()
         self._runtime_pool = OhmoSessionRuntimePool(
             cwd=self._cwd,

ohmo/workspace.py

@@ -283,6 +283,8 @@ def initialize_workspace(workspace: str | Path | None = None) -> Path:
                     "send_tool_hints": True,
                     "permission_mode": "default",
                     "sandbox_enabled": False,
+                    "allow_remote_admin_commands": False,
+                    "allowed_remote_admin_commands": [],
                     "log_level": "INFO",
                     "channel_configs": {},
                 },

src/openharness/commands/registry.py

@@ -100,6 +100,8 @@ class SlashCommand:
     name: str
     description: str
     handler: CommandHandler
+    remote_invocable: bool = True
+    remote_admin_opt_in: bool = False
 
 
 class CommandRegistry:
@@ -375,9 +377,9 @@ def create_default_command_registry(
             return CommandResult(message="\n".join(path.name for path in memory_files))
         if action == "show" and rest:
             memory_dir = get_project_memory_dir(context.cwd)
-            path = memory_dir / rest
-            if not path.exists():
-                path = memory_dir / f"{rest}.md"
+            path = _resolve_memory_entry_path(memory_dir, rest)
+            if path is None:
+                return CommandResult(message="Memory entry path must stay within the project memory directory.")
             if not path.exists():
                 return CommandResult(message=f"Memory entry not found: {rest}")
             return CommandResult(message=path.read_text(encoding="utf-8"))
@@ -1560,8 +1562,24 @@ def create_default_command_registry(
     registry.register(SlashCommand("mcp", "Show MCP status", _mcp_handler))
     registry.register(SlashCommand("plugin", "Manage plugins", _plugin_handler))
     registry.register(SlashCommand("reload-plugins", "Reload plugin discovery for this workspace", _reload_plugins_handler))
-    registry.register(SlashCommand("permissions", "Show or update permission mode", _permissions_handler))
-    registry.register(SlashCommand("plan", "Toggle plan permission mode", _plan_handler))
+    registry.register(
+        SlashCommand(
+            "permissions",
+            "Show or update permission mode",
+            _permissions_handler,
+            remote_invocable=False,
+            remote_admin_opt_in=True,
+        )
+    )
+    registry.register(
+        SlashCommand(
+            "plan",
+            "Toggle plan permission mode",
+            _plan_handler,
+            remote_invocable=False,
+            remote_admin_opt_in=True,
+        )
+    )
     registry.register(SlashCommand("fast", "Show or update fast mode", _fast_handler))
     registry.register(SlashCommand("effort", "Show or update reasoning effort", _effort_handler))
     registry.register(SlashCommand("passes", "Show or update reasoning pass count", _passes_handler))
@@ -1618,3 +1636,28 @@ def create_default_command_registry(
             )
         )
     return registry
+
+
+def _resolve_memory_entry_path(memory_dir: Path, candidate: str) -> Path | None:
+    """Resolve a memory entry path while enforcing containment under ``memory_dir``."""
+
+    base = memory_dir.resolve()
+    resolved = _resolve_memory_candidate(base, candidate)
+    if resolved is not None and resolved.exists():
+        return resolved
+    fallback = _resolve_memory_candidate(base, f"{candidate}.md")
+    if fallback is not None:
+        return fallback
+    return None
+
+
+def _resolve_memory_candidate(memory_dir: Path, candidate: str) -> Path | None:
+    path = Path(candidate).expanduser()
+    if not path.is_absolute():
+        path = memory_dir / path
+    resolved = path.resolve()
+    try:
+        resolved.relative_to(memory_dir)
+    except ValueError:
+        return None
+    return resolved

tests/test_commands/test_registry.py

@@ -75,6 +75,53 @@ async def test_permissions_command_persists(tmp_path: Path, monkeypatch):
     assert load_settings().permission.mode == "full_auto"
 
 
+@pytest.mark.asyncio
+async def test_permissions_command_is_marked_local_only(tmp_path: Path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    registry = create_default_command_registry()
+    command, _ = registry.lookup("/permissions set full_auto")
+    assert command is not None
+    assert command.remote_invocable is False
+
+
+@pytest.mark.asyncio
+async def test_permissions_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    registry = create_default_command_registry()
+    command, _ = registry.lookup("/permissions set full_auto")
+    assert command is not None
+    assert getattr(command, "remote_admin_opt_in", False) is True
+
+
+@pytest.mark.asyncio
+async def test_memory_show_rejects_path_traversal(tmp_path: Path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    registry = create_default_command_registry()
+    command, args = registry.lookup("/memory show ../../../../../../etc/hosts")
+    assert command is not None
+
+    result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path)))
+
+    assert result.message == "Memory entry path must stay within the project memory directory."
+
+
+@pytest.mark.asyncio
+async def test_memory_show_reads_normal_entries_with_md_fallback(tmp_path: Path, monkeypatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    registry = create_default_command_registry()
+
+    add_command, add_args = registry.lookup("/memory add Notes :: hello world")
+    assert add_command is not None
+    await add_command.handler(add_args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path)))
+
+    show_command, show_args = registry.lookup("/memory show Notes")
+    result = await show_command.handler(show_args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path)))
+
+    assert "hello world" in result.message
+
+
 @pytest.mark.asyncio
 async def test_model_command_persists(tmp_path: Path, monkeypatch):
     monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))

tests/test_ohmo/test_gateway.py

@@ -12,11 +12,13 @@ from openharness.api.usage import UsageSnapshot
 from openharness.channels.bus.events import InboundMessage
 from openharness.channels.bus.queue import MessageBus
 from openharness.commands import CommandResult
+from openharness.commands.registry import SlashCommand
 from openharness.engine.messages import ConversationMessage, ImageBlock, TextBlock
 from openharness.engine.stream_events import AssistantTextDelta, CompactProgressEvent, ToolExecutionStarted
 
 from ohmo.gateway.bridge import OhmoGatewayBridge, _format_gateway_error
-from ohmo.gateway.models import GatewayState
+from ohmo.gateway.config import save_gateway_config
+from ohmo.gateway.models import GatewayConfig, GatewayState
 from ohmo.gateway.runtime import OhmoSessionRuntimePool, _build_inbound_user_message, _format_channel_progress
 from ohmo.gateway.service import OhmoGatewayService, gateway_status, stop_gateway_process
 from ohmo.gateway.router import session_key_for_message
@@ -356,6 +358,121 @@ async def test_runtime_pool_stream_message_uses_english_progress_for_english_inp
     assert updates[1].text.startswith("🛠️ Using web_fetch")
 
 
+@pytest.mark.asyncio
+async def test_runtime_pool_blocks_local_only_commands_from_remote_messages(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    handler_called = False
+
+    async def forbidden_handler(args, context):
+        nonlocal handler_called
+        handler_called = True
+        return CommandResult(message="should not run")
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+        command = SlashCommand(
+            "permissions",
+            "Show or update permission mode",
+            forbidden_handler,
+            remote_invocable=False,
+        )
+        command.remote_admin_opt_in = True
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: (command, "full_auto")),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="/permissions full_auto")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert handler_called is False
+    assert updates[-1].kind == "final"
+    assert updates[-1].text == "/permissions is only available in the local OpenHarness UI."
+
+
+@pytest.mark.asyncio
+async def test_runtime_pool_allows_opted_in_remote_admin_commands(tmp_path, monkeypatch, caplog):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+    save_gateway_config(
+        GatewayConfig(
+            provider_profile="codex",
+            allow_remote_admin_commands=True,
+            allowed_remote_admin_commands=["permissions"],
+        ),
+        workspace,
+    )
+    handler_called = False
+
+    async def allowed_handler(args, context):
+        nonlocal handler_called
+        handler_called = True
+        return CommandResult(message=f"ran with {args}")
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+        command = SlashCommand(
+            "permissions",
+            "Show or update permission mode",
+            allowed_handler,
+            remote_invocable=False,
+        )
+        command.remote_admin_opt_in = True
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: (command, "full_auto")),
+            hook_summary=lambda: "",
+            mcp_summary=lambda: "",
+            plugin_summary=lambda: "",
+            cwd=str(tmp_path),
+            tool_registry=None,
+            app_state=None,
+            session_backend=None,
+            extra_skill_dirs=(),
+            extra_plugin_roots=(),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    with caplog.at_level(logging.WARNING):
+        pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+        message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="/permissions full_auto")
+        updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert handler_called is True
+    assert updates[-1].kind == "final"
+    assert updates[-1].text == "ran with full_auto"
+    assert "remote administrative command accepted" in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_runtime_pool_includes_media_paths_in_prompt(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"