fix(security): route LobeHub skills fetches through guarded HTTP

tests/tools/test_lobehub_skills_http_ssrf.py

@@ -0,0 +1,32 @@
+"""LobeHub Skills Hub fetch uses guarded HTTP + agent-id sanitization."""
+
+from unittest.mock import MagicMock, patch
+
+from tools.skills_hub import LobeHubSource
+
+
+def test_lobehub_sanitize_rejects_traversal():
+    src = LobeHubSource()
+    assert src._sanitize_agent_id("../etc/passwd") is None
+    assert src._sanitize_agent_id("http://evil.example/x") is None
+    assert src._sanitize_agent_id("a/b") is None
+    assert src._sanitize_agent_id("safe-agent_1") == "safe-agent_1"
+
+
+def test_lobehub_fetch_index_uses_guarded_http():
+    src = LobeHubSource()
+    fake = MagicMock()
+    fake.status_code = 200
+    fake.json.return_value = {"agents": []}
+    with patch("tools.skills_hub._read_index_cache", return_value=None), patch(
+        "tools.skills_hub._write_index_cache"
+    ), patch("tools.skills_hub._guarded_http_get", return_value=fake) as guarded:
+        assert src._fetch_index() == {"agents": []}
+        guarded.assert_called_once_with(src.INDEX_URL, timeout=30)
+
+
+def test_lobehub_fetch_agent_rejects_unsafe_id():
+    src = LobeHubSource()
+    with patch("tools.skills_hub._guarded_http_get") as guarded:
+        assert src._fetch_agent("../../x") is None
+        guarded.assert_not_called()

tests/tools/test_skills_hub.py

@@ -1470,6 +1470,26 @@ def test_list_taps(self, tmp_path):
         assert len(taps) == 2
 
 
+# ---------------------------------------------------------------------------
+# LobeHubSource._fetch_agent
+# ---------------------------------------------------------------------------
+
+
+class TestFetchAgent:
+    @patch("tools.skills_hub._guarded_http_get")
+    def test_valid_id_uses_guarded_fetch_with_agent_url_and_timeout(self, mock_get):
+        response = MagicMock(status_code=200)
+        response.json.return_value = {"identifier": "test-agent"}
+        mock_get.return_value = response
+
+        result = LobeHubSource()._fetch_agent("test-agent")
+
+        assert result == {"identifier": "test-agent"}
+        mock_get.assert_called_once_with(
+            "https://chat-agents.lobehub.com/test-agent.json", timeout=15
+        )
+
+
 # ---------------------------------------------------------------------------
 # LobeHubSource._convert_to_skill_md
 # ---------------------------------------------------------------------------

tools/skills_hub.py

@@ -2922,34 +2922,51 @@ def inspect(self, identifier: str) -> Optional[SkillMeta]:
                 )
         return None
 
+    @staticmethod
+    def _sanitize_agent_id(agent_id: str) -> Optional[str]:
+        """Reject path traversal / absolute / scheme-bearing agent ids."""
+        cleaned = (agent_id or "").strip()
+        if not cleaned or len(cleaned) > 200:
+            return None
+        if any(ch in cleaned for ch in ("/", "\\", ":", "?", "#", "@")):
+            return None
+        if ".." in cleaned or cleaned.startswith("."):
+            return None
+        return cleaned
+
     def _fetch_index(self) -> Optional[Any]:
         """Fetch the LobeHub agent index (cached for 1 hour)."""
         cache_key = "lobehub_index"
         cached = _read_index_cache(cache_key)
         if cached is not None:
             return cached
 
+        resp = _guarded_http_get(self.INDEX_URL, timeout=30)
+        if resp is None or resp.status_code != 200:
+            return None
         try:
-            resp = httpx.get(self.INDEX_URL, timeout=30)
-            if resp.status_code != 200:
-                return None
             data = resp.json()
-        except (httpx.HTTPError, json.JSONDecodeError):
+        except (ValueError, json.JSONDecodeError):
             return None
 
         _write_index_cache(cache_key, data)
         return data
 
     def _fetch_agent(self, agent_id: str) -> Optional[dict]:
-        """Fetch a single agent's JSON file."""
-        url = f"https://chat-agents.lobehub.com/{agent_id}.json"
+        """Fetch a single agent's JSON file via the guarded Skills Hub HTTP path."""
+        safe_id = self._sanitize_agent_id(agent_id)
+        if not safe_id:
+            logger.warning("LobeHub: rejected unsafe agent id %r", agent_id)
+            return None
+        url = f"https://chat-agents.lobehub.com/{safe_id}.json"
+        resp = _guarded_http_get(url, timeout=15)
+        if resp is None or resp.status_code != 200:
+            return None
         try:
-            resp = httpx.get(url, timeout=15)
-            if resp.status_code == 200:
-                return resp.json()
-        except (httpx.HTTPError, json.JSONDecodeError) as e:
+            return resp.json()
+        except (ValueError, json.JSONDecodeError) as e:
             logger.debug("LobeHub agent fetch failed: %s", e)
-        return None
+            return None
 
     @staticmethod
     def _convert_to_skill_md(agent_data: dict) -> str: