fix: harden gateway slash command security

ohmo/cli.py

@@ -306,13 +306,31 @@ 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,
             "enabled_channels": enabled_channels,
             "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 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.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,13 +82,26 @@ def __init__(
         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] = {}
 
     @property
     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 @@ async def stream_message(self, message: InboundMessage, session_key: str):
         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 @@ def __init__(self, cwd: str | Path | None = None, workspace: str | Path | None =
         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 @@ async def _memory_handler(args: str, context: CommandContext) -> CommandResult:
             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"))
@@ -1545,8 +1547,24 @@ async def _tasks_handler(args: str, context: CommandContext) -> CommandResult:
     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))
@@ -1602,3 +1620,28 @@ async def _plugin_command_handler(
             )
         )
     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"))