From 8b395fbab2f0fba30f958d65bed6d7ce984215c5 Mon Sep 17 00:00:00 2001 From: SEPURI-SAI-KRISHNA Date: Tue, 4 Aug 2026 12:54:55 +0530 Subject: [PATCH] fix(v2ex): percent-encode caller values in API query strings --- agent_reach/channels/v2ex.py | 41 +++++++++---- tests/test_v2ex_channel.py | 115 +++++++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 11 deletions(-) diff --git a/agent_reach/channels/v2ex.py b/agent_reach/channels/v2ex.py index b6032522..4af3bff2 100644 --- a/agent_reach/channels/v2ex.py +++ b/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 搜索。" ) } diff --git a/tests/test_v2ex_channel.py b/tests/test_v2ex_channel.py index 2379ed62..86f29345 100644 --- a/tests/test_v2ex_channel.py +++ b/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"