fix: OpenClaw ports — weak credential guard + Matrix m.mentions

gateway/config.py

@@ -665,6 +665,17 @@ def load_gateway_config() -> GatewayConfig:
     _apply_env_overrides(config)
 
     # --- Validate loaded values ---
+    _validate_gateway_config(config)
+
+    return config
+
+
+def _validate_gateway_config(config: "GatewayConfig") -> None:
+    """Validate and sanitize a loaded GatewayConfig in place.
+
+    Called by ``load_gateway_config()`` after all config sources are merged.
+    Extracted as a separate function for testability.
+    """
     policy = config.default_reset_policy
 
     if not (0 <= policy.at_hour <= 23):
@@ -701,7 +712,31 @@ def load_gateway_config() -> GatewayConfig:
                 platform.value, env_name,
             )
 
-    return config
+    # Reject known-weak placeholder tokens.
+    # Ported from openclaw/openclaw#64586: users who copy .env.example
+    # without changing placeholder values get a clear startup error instead
+    # of a confusing "auth failed" from the platform API.
+    try:
+        from hermes_cli.auth import has_usable_secret
+    except ImportError:
+        has_usable_secret = None  # type: ignore[assignment]
+
+    if has_usable_secret is not None:
+        for platform, pconfig in config.platforms.items():
+            if not pconfig.enabled:
+                continue
+            env_name = _token_env_names.get(platform)
+            if not env_name:
+                continue
+            token = pconfig.token
+            if token and token.strip() and not has_usable_secret(token, min_length=4):
+                logger.error(
+                    "%s is enabled but %s is set to a placeholder value ('%s'). "
+                    "Set a real bot token before starting the gateway. "
+                    "The adapter will NOT be started.",
+                    platform.value, env_name, token.strip()[:6] + "...",
+                )
+                pconfig.enabled = False
 
 
 def _apply_env_overrides(config: GatewayConfig) -> None:

gateway/platforms/api_server.py

@@ -1819,6 +1819,23 @@ async def connect(self) -> bool:
                 )
                 return False
 
+            # Refuse to start network-accessible with a placeholder key.
+            # Ported from openclaw/openclaw#64586.
+            if is_network_accessible(self._host) and self._api_key:
+                try:
+                    from hermes_cli.auth import has_usable_secret
+                    if not has_usable_secret(self._api_key, min_length=8):
+                        logger.error(
+                            "[%s] Refusing to start: API_SERVER_KEY is set to a "
+                            "placeholder value. Generate a real secret "
+                            "(e.g. `openssl rand -hex 32`) and set API_SERVER_KEY "
+                            "before exposing the API server on %s.",
+                            self.name, self._host,
+                        )
+                        return False
+                except ImportError:
+                    pass
+
             # Port conflict detection — fail fast if port is already in use
             try:
                 with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as _s:

gateway/platforms/matrix.py

@@ -1135,7 +1135,10 @@ async def _resolve_message_context(
             thread_id = relates_to.get("event_id")
 
         formatted_body = source_content.get("formatted_body")
-        is_mentioned = self._is_bot_mentioned(body, formatted_body)
+        # m.mentions.user_ids (MSC3952 / Matrix v1.7) — authoritative mention signal.
+        mentions_block = source_content.get("m.mentions") or {}
+        mention_user_ids = mentions_block.get("user_ids") if isinstance(mentions_block, dict) else None
+        is_mentioned = self._is_bot_mentioned(body, formatted_body, mention_user_ids)
 
         # Require-mention gating.
         if not is_dm:
@@ -1822,8 +1825,24 @@ async def _refresh_dm_cache(self) -> None:
     # Mention detection helpers
     # ------------------------------------------------------------------
 
-    def _is_bot_mentioned(self, body: str, formatted_body: Optional[str] = None) -> bool:
-        """Return True if the bot is mentioned in the message."""
+    def _is_bot_mentioned(
+        self,
+        body: str,
+        formatted_body: Optional[str] = None,
+        mention_user_ids: Optional[list] = None,
+    ) -> bool:
+        """Return True if the bot is mentioned in the message.
+
+        Per MSC3952, ``m.mentions.user_ids`` is the authoritative mention
+        signal in the Matrix spec.  When the sender's client populates that
+        field with the bot's user-id, we trust it — even when the visible
+        body text does not contain an explicit ``@bot`` string (some clients
+        only render mention "pills" in ``formatted_body`` or use display
+        names).
+        """
+        # m.mentions.user_ids — authoritative per MSC3952 / Matrix v1.7.
+        if mention_user_ids and self._user_id and self._user_id in mention_user_ids:
+            return True
         if not body and not formatted_body:
             return False
         if self._user_id and self._user_id in body:

tests/gateway/test_matrix_mention.py

@@ -48,6 +48,7 @@ def _make_event(
     room_id="!room1:example.org",
     formatted_body=None,
     thread_id=None,
+    mention_user_ids=None,
 ):
     """Create a fake room message event.
 
@@ -60,6 +61,9 @@ def _make_event(
         content["formatted_body"] = formatted_body
         content["format"] = "org.matrix.custom.html"
 
+    if mention_user_ids is not None:
+        content["m.mentions"] = {"user_ids": mention_user_ids}
+
     relates_to = {}
     if thread_id:
         relates_to["rel_type"] = "m.thread"
@@ -108,6 +112,44 @@ def test_partial_localpart_no_match(self):
         # "hermesbot" should not match word-boundary check for "hermes"
         assert not self.adapter._is_bot_mentioned("hermesbot is here")
 
+    # m.mentions.user_ids — MSC3952 / Matrix v1.7 authoritative mentions
+    # Ported from openclaw/openclaw#64796
+
+    def test_m_mentions_user_ids_authoritative(self):
+        """m.mentions.user_ids alone is sufficient — no body text needed."""
+        assert self.adapter._is_bot_mentioned(
+            "please reply",  # no @hermes anywhere in body
+            mention_user_ids=["@hermes:example.org"],
+        )
+
+    def test_m_mentions_user_ids_with_body_mention(self):
+        """Both m.mentions and body mention — should still be True."""
+        assert self.adapter._is_bot_mentioned(
+            "hey @hermes:example.org help",
+            mention_user_ids=["@hermes:example.org"],
+        )
+
+    def test_m_mentions_user_ids_other_user_only(self):
+        """m.mentions with a different user — bot is NOT mentioned."""
+        assert not self.adapter._is_bot_mentioned(
+            "hello",
+            mention_user_ids=["@alice:example.org"],
+        )
+
+    def test_m_mentions_user_ids_empty_list(self):
+        """Empty user_ids list — falls through to text detection."""
+        assert not self.adapter._is_bot_mentioned(
+            "hello everyone",
+            mention_user_ids=[],
+        )
+
+    def test_m_mentions_user_ids_none(self):
+        """None mention_user_ids — falls through to text detection."""
+        assert not self.adapter._is_bot_mentioned(
+            "hello everyone",
+            mention_user_ids=None,
+        )
+
 
 class TestStripMention:
     def setup_method(self):
@@ -176,6 +218,44 @@ async def test_require_mention_html_pill(monkeypatch):
     adapter.handle_message.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_require_mention_m_mentions_user_ids(monkeypatch):
+    """m.mentions.user_ids is authoritative per MSC3952 — no body mention needed.
+
+    Ported from openclaw/openclaw#64796.
+    """
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    # Body has NO mention, but m.mentions.user_ids includes the bot.
+    event = _make_event(
+        "please reply",
+        mention_user_ids=["@hermes:example.org"],
+    )
+
+    await adapter._on_room_message(event)
+    adapter.handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_require_mention_m_mentions_other_user_ignored(monkeypatch):
+    """m.mentions.user_ids mentioning another user should NOT activate the bot."""
+    monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+    monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+    adapter = _make_adapter()
+    event = _make_event(
+        "hey alice check this",
+        mention_user_ids=["@alice:example.org"],
+    )
+
+    await adapter._on_room_message(event)
+    adapter.handle_message.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_require_mention_dm_always_responds(monkeypatch):
     """DMs always respond regardless of mention setting."""

tests/gateway/test_weak_credential_guard.py

@@ -0,0 +1,141 @@
+"""Tests for gateway weak credential rejection at startup.
+
+Ported from openclaw/openclaw#64586: rejects known-weak placeholder
+tokens at gateway startup instead of letting them silently fail
+against platform APIs.
+"""
+
+import logging
+
+import pytest
+
+from gateway.config import PlatformConfig, Platform, _validate_gateway_config
+
+
+# ---------------------------------------------------------------------------
+# Helper: create a minimal GatewayConfig with one enabled platform
+# ---------------------------------------------------------------------------
+
+
+def _make_gateway_config(platform, token, enabled=True, **extra_kwargs):
+    """Create a minimal GatewayConfig-like object for validation testing."""
+    from gateway.config import GatewayConfig
+
+    config = GatewayConfig(platforms={})
+    pconfig = PlatformConfig(enabled=enabled, token=token, **extra_kwargs)
+    config.platforms[platform] = pconfig
+    return config
+
+
+def _validate_and_return(config):
+    """Call _validate_gateway_config and return the config (mutated in place)."""
+    _validate_gateway_config(config)
+    return config
+
+
+# ---------------------------------------------------------------------------
+# Unit tests: platform token placeholder rejection
+# ---------------------------------------------------------------------------
+
+
+class TestPlatformTokenPlaceholderGuard:
+    """Verify that _validate_gateway_config disables platforms with placeholder tokens."""
+
+    def test_rejects_triple_asterisk(self, caplog):
+        """'***' is the .env.example placeholder — should be rejected."""
+        config = _make_gateway_config(Platform.TELEGRAM, "***")
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.TELEGRAM].enabled is False
+        assert "placeholder" in caplog.text.lower()
+
+    def test_rejects_changeme(self, caplog):
+        config = _make_gateway_config(Platform.DISCORD, "changeme")
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.DISCORD].enabled is False
+
+    def test_rejects_your_api_key(self, caplog):
+        config = _make_gateway_config(Platform.SLACK, "your_api_key")
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.SLACK].enabled is False
+
+    def test_rejects_placeholder(self, caplog):
+        config = _make_gateway_config(Platform.MATRIX, "placeholder")
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.MATRIX].enabled is False
+
+    def test_accepts_real_token(self, caplog):
+        """A real-looking bot token should pass validation."""
+        config = _make_gateway_config(
+            Platform.TELEGRAM, "7123456789:AAHdqTcvCH1vGWJxfSeOfSAs0K5PALDsaw"
+        )
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.TELEGRAM].enabled is True
+        assert "placeholder" not in caplog.text.lower()
+
+    def test_accepts_empty_token_without_error(self, caplog):
+        """Empty tokens get a warning (existing behavior), not a placeholder error."""
+        config = _make_gateway_config(Platform.TELEGRAM, "")
+        with caplog.at_level(logging.WARNING):
+            _validate_and_return(config)
+        # Empty token doesn't trigger placeholder rejection — enabled stays True
+        # (the existing empty-token warning is separate)
+        assert config.platforms[Platform.TELEGRAM].enabled is True
+
+    def test_disabled_platform_not_checked(self, caplog):
+        """Disabled platforms should not be validated."""
+        config = _make_gateway_config(Platform.TELEGRAM, "***", enabled=False)
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert "placeholder" not in caplog.text.lower()
+
+    def test_rejects_whitespace_padded_placeholder(self, caplog):
+        """Whitespace-padded placeholders should still be caught."""
+        config = _make_gateway_config(Platform.TELEGRAM, "  ***  ")
+        with caplog.at_level(logging.ERROR):
+            _validate_and_return(config)
+        assert config.platforms[Platform.TELEGRAM].enabled is False
+
+
+# ---------------------------------------------------------------------------
+# Integration test: API server placeholder key on network-accessible host
+# ---------------------------------------------------------------------------
+
+
+class TestAPIServerPlaceholderKeyGuard:
+    """Verify that the API server rejects placeholder keys on network hosts."""
+
+    @pytest.mark.asyncio
+    async def test_refuses_wildcard_with_placeholder_key(self):
+        from gateway.platforms.api_server import APIServerAdapter
+
+        adapter = APIServerAdapter(
+            PlatformConfig(enabled=True, extra={"host": "0.0.0.0", "key": "changeme"})
+        )
+        result = await adapter.connect()
+        assert result is False
+
+    @pytest.mark.asyncio
+    async def test_refuses_wildcard_with_asterisk_key(self):
+        from gateway.platforms.api_server import APIServerAdapter
+
+        adapter = APIServerAdapter(
+            PlatformConfig(enabled=True, extra={"host": "0.0.0.0", "key": "***"})
+        )
+        result = await adapter.connect()
+        assert result is False
+
+    def test_allows_loopback_with_placeholder_key(self):
+        """Loopback with a placeholder key is fine — not network-exposed."""
+        from gateway.platforms.api_server import APIServerAdapter
+        from gateway.platforms.base import is_network_accessible
+
+        adapter = APIServerAdapter(
+            PlatformConfig(enabled=True, extra={"host": "127.0.0.1", "key": "changeme"})
+        )
+        # On loopback the placeholder guard doesn't fire
+        assert is_network_accessible(adapter._host) is False