fix(v2ex): percent-encode caller values in API query strings

agent_reach/channels/v2ex.py

@@ -2,6 +2,7 @@
 """V2EX — public API channel for topics, nodes, users, and replies."""
 
 import json
+import urllib.parse
 import urllib.request
 from typing import Any
 
@@ -11,6 +12,19 @@
 
 _UA = "agent-reach/1.0"
 _TIMEOUT = 10
+_API_BASE = "https://www.v2ex.com"
+
+
+def _api_url(path: str, **params: Any) -> str:
+    """Build a V2EX API URL with every caller value percent-encoded.
+
+    Node names, usernames and topic ids arrive from agents and from parsed
+    URLs. Interpolating them raw lets ``&`` add or override a parameter, ``#``
+    drop the rest of the query into a fragment that is never sent, ``+`` decode
+    server-side as a space, and any non-ASCII value raise UnicodeEncodeError
+    out of urllib instead of a channel error.
+    """
+    return f"{_API_BASE}{path}?{urllib.parse.urlencode(params)}"
 
 
 def _get_json(url: str) -> Any:
@@ -92,10 +106,7 @@ def get_node_topics(self, node_name: str, limit: int = 20) -> list:
         Returns a list of dicts with keys:
           title, url, replies, node_name, node_title, content
         """
-        url = (
-            f"https://www.v2ex.com/api/topics/show.json"
-            f"?node_name={node_name}&page=1"
-        )
+        url = _api_url("/api/topics/show.json", node_name=node_name, page=1)
         data = _get_json(url)
         results = []
         for item in data[:limit]:
@@ -126,7 +137,7 @@ def get_topic(self, topic_id: int) -> dict:
           author, created, replies (list of dicts with: author, content, created)
         """
         topic_data = _get_json(
-            f"https://www.v2ex.com/api/topics/show.json?id={topic_id}"
+            _api_url("/api/topics/show.json", id=topic_id)
         )
         # API returns a list even for single-ID queries
         if isinstance(topic_data, list):
@@ -140,8 +151,9 @@ def get_topic(self, topic_id: int) -> dict:
         # Fetch replies (first page)
         try:
             replies_raw = _get_json(
-                f"https://www.v2ex.com/api/replies/show.json"
-                f"?topic_id={topic_id}&page=1"
+                _api_url(
+                    "/api/replies/show.json", topic_id=topic_id, page=1
+                )
             )
         except Exception:
             replies_raw = []
@@ -158,7 +170,10 @@ def get_topic(self, topic_id: int) -> dict:
         return {
             "id": topic.get("id", topic_id),
             "title": topic.get("title", ""),
-            "url": topic.get("url", f"https://www.v2ex.com/t/{topic_id}"),
+            "url": topic.get(
+                "url",
+                f"{_API_BASE}/t/{urllib.parse.quote(str(topic_id), safe='')}",
+            ),
             "content": topic.get("content", ""),
             "replies_count": topic.get("replies", 0),
             "node_name": node.get("name", ""),
@@ -179,12 +194,15 @@ def get_user(self, username: str) -> dict:
           location, bio, avatar, created
         """
         data = _get_json(
-            f"https://www.v2ex.com/api/members/show.json?username={username}"
+            _api_url("/api/members/show.json", username=username)
         )
         return {
             "id": data.get("id", 0),
             "username": data.get("username", username),
-            "url": data.get("url", f"https://www.v2ex.com/member/{username}"),
+            "url": data.get(
+                "url",
+                f"{_API_BASE}/member/{urllib.parse.quote(str(username), safe='')}",
+            ),
             "website": data.get("website", ""),
             "twitter": data.get("twitter", ""),
             "psn": data.get("psn", ""),
@@ -209,11 +227,12 @@ def search(self, query: str, limit: int = 10) -> list:
             list of dicts with keys: title, url, snippet
             如果搜索不可用，返回包含单条 {"error": str} 的列表。
         """
+        search_url = _api_url("/", q=query)
         return [
             {
                 "error": (
                     "V2EX 公开 API 不提供搜索端点。"
-                    f"建议改用：https://www.v2ex.com/?q={query} "
+                    f"建议改用：{search_url} "
                     "或通过 Exa channel 使用 site:v2ex.com 搜索。"
                 )
             }

tests/test_v2ex_channel.py

@@ -12,6 +12,9 @@
 """
 
 from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
 
 from agent_reach.channels import v2ex as v2
 from agent_reach.channels.v2ex import V2EXChannel
@@ -153,3 +156,115 @@ def test_search_returns_guidance_without_network():
     assert len(results) == 1
     assert "error" in results[0]
     assert "python" in results[0]["error"]
+
+
+# --- query-string encoding ---
+#
+# Node names, usernames and topic ids reach these methods from agents and from
+# parsed URLs. Interpolated raw, "&" adds a parameter, "#" drops the rest of
+# the query into a never-sent fragment, "+" decodes server-side as a space, and
+# non-ASCII raises UnicodeEncodeError out of urllib instead of a channel error.
+
+_HOSTILE_VALUES = [
+    "python&page=99",   # would inject a duplicate page parameter
+    "foo#bar",          # would truncate the query into a fragment
+    "c++",              # would decode server-side as "c  "
+    "hello world",      # raw space in a URL
+    "Python 开发",       # would raise UnicodeEncodeError in urllib
+]
+
+
+def _captured_urls(call):
+    seen = []
+
+    def fake_get_json(url):
+        seen.append(url)
+        raise _StopFetch
+
+    with patch.object(v2, "_get_json", fake_get_json):
+        try:
+            call()
+        except _StopFetch:
+            pass
+    return seen
+
+
+class _StopFetch(Exception):
+    """Abort a channel method once the URL has been captured."""
+
+
+@pytest.mark.parametrize("value", _HOSTILE_VALUES)
+def test_get_node_topics_percent_encodes_node_name(value):
+    ch = V2EXChannel()
+    url = _captured_urls(lambda: ch.get_node_topics(value))[0]
+    parts = urlsplit(url)
+    query = parse_qs(parts.query)
+
+    assert parts.netloc == "www.v2ex.com"
+    assert parts.fragment == ""
+    assert query["node_name"] == [value]
+    assert query["page"] == ["1"]
+
+
+@pytest.mark.parametrize("value", _HOSTILE_VALUES)
+def test_get_user_percent_encodes_username(value):
+    ch = V2EXChannel()
+    url = _captured_urls(lambda: ch.get_user(value))[0]
+    parts = urlsplit(url)
+
+    assert parts.fragment == ""
+    assert parse_qs(parts.query)["username"] == [value]
+
+
+def test_get_topic_percent_encodes_ids_in_both_requests():
+    ch = V2EXChannel()
+    # Typed as int, but nothing enforces it — a stray "#" must not truncate.
+    urls = _captured_urls(lambda: ch.get_topic("1#"))
+
+    topic_query = parse_qs(urlsplit(urls[0]).query)
+    assert urlsplit(urls[0]).fragment == ""
+    assert topic_query["id"] == ["1#"]
+
+
+def test_get_topic_replies_request_keeps_page_parameter():
+    ch = V2EXChannel()
+    seen = []
+
+    def fake_get_json(url):
+        seen.append(url)
+        return [] if len(seen) > 1 else [{"id": 1}]
+
+    with patch.object(v2, "_get_json", fake_get_json):
+        ch.get_topic("1#")
+
+    replies_parts = urlsplit(seen[1])
+    replies_query = parse_qs(replies_parts.query)
+    assert replies_parts.fragment == ""
+    assert replies_query["topic_id"] == ["1#"]
+    assert replies_query["page"] == ["1"]
+
+
+def test_get_topic_accepts_plain_int_unchanged():
+    ch = V2EXChannel()
+    url = _captured_urls(lambda: ch.get_topic(123))[0]
+    assert parse_qs(urlsplit(url).query)["id"] == ["123"]
+
+
+def test_search_advisory_url_is_encoded():
+    ch = V2EXChannel()
+    with patch.object(v2, "_get_json", side_effect=AssertionError("must not hit network")):
+        message = ch.search("rust & go")[0]["error"]
+
+    assert "?q=rust+%26+go" in message
+    assert "?q=rust & go" not in message
+
+
+def test_fallback_display_urls_are_encoded():
+    ch = V2EXChannel()
+    with patch.object(v2, "_get_json", return_value={}):
+        user_url = ch.get_user("a b/c")["url"]
+    with patch.object(v2, "_get_json", return_value={}):
+        topic_url = ch.get_topic("9 9")["url"]
+
+    assert user_url == "https://www.v2ex.com/member/a%20b%2Fc"
+    assert topic_url == "https://www.v2ex.com/t/9%209"