fix(review-feedback-1708): 落地可配置 RSS/Atom 情报源、存储、查询、retention 和基础安全边界

Author: ZhuLinsen

docs/intelligence-sources.md

@@ -18,7 +18,8 @@ Issue #1707 的首版能力聚焦“合规资讯源采集、本地沉淀、可
 
 - 只允许绝对 `http` / `https` URL；
 - 禁止 URL 中携带 username/password；
-- 禁止 `localhost`、`.local`、回环地址、内网地址、链路本地地址、保留地址和组播地址；
+- 禁止 `localhost`、`.local`、回环地址、内网地址、链路本地地址、保留地址、共享地址段和组播地址；
+- 解析与拉取阶段显式禁用环境代理（如 `HTTP_PROXY`、`HTTPS_PROXY`、`ALL_PROXY`），避免通过环境代理绕过校验边界；
 - 实际连接阶段会再次校验目标主机 DNS 解析结果，避免校验后解析漂移到受限地址；
 - 重定向后的最终 URL 也会再次校验；
 - 错误消息会脱敏常见 `token` / `key` / `secret` 查询参数。

src/services/intelligence_service.py

@@ -32,6 +32,7 @@
 _MAX_REDIRECTS = 5
 _REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
 _DNS_GUARD_LOCK = threading.RLock()
+_DISABLE_REQUEST_PROXIES = {"http": None, "https": None}
 
 
 class IntelligenceServiceError(ValueError):
@@ -248,7 +249,9 @@ def guarded_getaddrinfo(host: Any, port: Any, *args: Any, **inner_kwargs: Any) -
         with _DNS_GUARD_LOCK:
             socket.getaddrinfo = guarded_getaddrinfo
             try:
-                return requests.get(raw_url, **kwargs)
+                request_kwargs = dict(kwargs)
+                request_kwargs.setdefault("proxies", _DISABLE_REQUEST_PROXIES)
+                return requests.get(raw_url, **request_kwargs)
             finally:
                 socket.getaddrinfo = original_getaddrinfo
 
@@ -309,18 +312,11 @@ def _validate_addrinfos(addrinfos: Any) -> None:
             except ValueError as exc:
                 raise IntelligenceServiceError("source url resolved to an invalid address") from exc
             if IntelligenceService._is_blocked_ip(ip):
-                raise IntelligenceServiceError("source url must not target private or local network addresses")
+                raise IntelligenceServiceError("source url must resolve to a public internet address")
 
     @staticmethod
     def _is_blocked_ip(ip: ipaddress._BaseAddress) -> bool:
-        return (
-            ip.is_private
-            or ip.is_loopback
-            or ip.is_link_local
-            or ip.is_reserved
-            or ip.is_multicast
-            or ip.is_unspecified
-        )
+        return not ip.is_global
 
     def _parse_feed(self, content: bytes, *, source_name: str, limit: int) -> List[FeedEntry]:
         try:

tests/test_intelligence_service.py

@@ -104,6 +104,12 @@ def test_dns_name_resolving_private_address_is_rejected(self) -> None:
             with self.assertRaises(IntelligenceServiceError):
                 self.service.create_source({"name": "bad", "url": "https://metadata.example.com/rss.xml", "scope_type": "market"})
 
+    def test_shared_address_space_url_is_rejected(self) -> None:
+        shared_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("100.64.0.1", 0))]
+        with patch("src.services.intelligence_service.socket.getaddrinfo", return_value=shared_dns):
+            with self.assertRaises(IntelligenceServiceError):
+                self.service.create_source({"name": "bad", "url": "https://metadata.example.com/rss.xml", "scope_type": "market"})
+
     def test_generated_no_url_sentinel_is_rejected_for_source_url(self) -> None:
         with self.assertRaisesRegex(IntelligenceServiceError, "absolute http\\(s\\) URL"):
             self.service.create_source({"name": "bad", "url": "no-url:intel:anything", "scope_type": "market"})
@@ -124,8 +130,21 @@ def test_redirect_target_is_validated_before_following(self) -> None:
         with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=redirect) as mock_get:
             with self.assertRaises(IntelligenceServiceError):
                 self.service.fetch_source(source["id"])
+            self.assertEqual(mock_get.call_count, 1)
+            self.assertFalse(mock_get.call_args.kwargs["allow_redirects"])
+
+    def test_fetch_requests_disable_environment_proxies(self) -> None:
+        with self._public_dns():
+            source = self.service.create_source({
+                "name": "proxy-safe-feed",
+                "url": "https://feeds.example.com/rss.xml",
+                "scope_type": "market",
+            })
+        with patch.dict(os.environ, {"HTTP_PROXY": "http://127.0.0.1:3128", "HTTPS_PROXY": "http://127.0.0.1:3128", "ALL_PROXY": "http://127.0.0.1:3128"}):
+            with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()) as mock_get:
+                self.service.fetch_source(source["id"])
         self.assertEqual(mock_get.call_count, 1)
-        self.assertFalse(mock_get.call_args.kwargs["allow_redirects"])
+        self.assertEqual(mock_get.call_args.kwargs["proxies"], {"http": None, "https": None})
 
     def test_fetch_validates_dns_resolution_used_by_request(self) -> None:
         public_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]