fix: bound PyPI metadata response reads

Author: bluecloud-gilfoyle[bot]

src/mcp_server_python_docs/services/package_docs.py

@@ -20,10 +20,11 @@
     "repository", "repo", "bug tracker", "issues", "changelog", "release notes",
 }
 _BLOCKED = ("mirror", "community", "unofficial", "tutorial", "example")
+_PYPI_METADATA_MAX_BYTES = 5 * 1024 * 1024
 
 
 class _HTTPResponse(Protocol):
-    def read(self) -> bytes: ...
+    def read(self, size: int = -1) -> bytes: ...
     def __enter__(self) -> "_HTTPResponse": ...
     def __exit__(self, exc_type: object, exc: object, tb: object) -> bool | None: ...
 
@@ -57,6 +58,13 @@ def _source(label: str, url: object, kind: str) -> PackageDocsSource | None:
     return PackageDocsSource(label=label, url=valid, kind=kind, declared_by="PyPI project metadata")
 
 
+def _read_limited(response: _HTTPResponse) -> bytes | None:
+    data = response.read(_PYPI_METADATA_MAX_BYTES + 1)
+    if len(data) > _PYPI_METADATA_MAX_BYTES:
+        return None
+    return data
+
+
 class PackageDocsService:
     """Return package-declared docs/homepage/source URLs from PyPI metadata only."""
 
@@ -70,7 +78,16 @@ def lookup(self, package: str) -> PackageDocsResult:
         metadata_source = f"https://pypi.org/pypi/{project}/json"
         try:
             with self._fetcher(metadata_source, self._timeout) as response:
-                payload = json.loads(response.read().decode("utf-8"))
+                data = _read_limited(response)
+                if data is None:
+                    return PackageDocsResult(
+                        package=package,
+                        version="",
+                        metadata_source=metadata_source,
+                        sources=[],
+                        note="PyPI metadata exceeded size limit.",
+                    )
+                payload = json.loads(data.decode("utf-8"))
         except HTTPError as e:
             if e.code == 404:
                 return PackageDocsResult(

tests/test_package_docs.py

@@ -2,20 +2,27 @@
 from __future__ import annotations
 
 import json
-from urllib.error import HTTPError
+from urllib.error import HTTPError, URLError
 
-from mcp_server_python_docs.services.package_docs import PackageDocsService
+from mcp_server_python_docs.services.package_docs import (
+    _PYPI_METADATA_MAX_BYTES,
+    PackageDocsService,
+)
 
 
 class _Resp:
-    def __init__(self, payload: dict):
+    def __init__(self, payload: dict | bytes):
         self._payload = payload
     def __enter__(self):
         return self
     def __exit__(self, exc_type, exc, tb):
         return False
-    def read(self) -> bytes:
-        return json.dumps(self._payload).encode()
+    def read(self, size: int = -1) -> bytes:
+        if isinstance(self._payload, bytes):
+            data = self._payload
+        else:
+            data = json.dumps(self._payload).encode()
+        return data if size < 0 else data[:size]
 
 
 def test_package_docs_uses_official_pypi_metadata_and_declared_urls():
@@ -67,3 +74,46 @@ def missing(url: str, timeout: float):
     tool = create_server()._tool_manager._tools["lookup_package_docs"]
     assert tool.annotations.readOnlyHint is True
     assert tool.annotations.openWorldHint is True
+
+
+def test_package_docs_rejects_oversized_pypi_metadata_without_unbounded_read():
+    class LargeResp:
+        requested_size: int | None = None
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self, size: int = -1) -> bytes:
+            self.requested_size = size
+            assert size == _PYPI_METADATA_MAX_BYTES + 1
+            return b"x" * size
+
+    response = LargeResp()
+
+    def fetch(url: str, timeout: float):
+        return response
+
+    result = PackageDocsService(fetcher=fetch).lookup("huge-package")
+
+    assert response.requested_size == _PYPI_METADATA_MAX_BYTES + 1
+    assert result.sources == []
+    assert result.note == "PyPI metadata exceeded size limit."
+
+
+def test_package_docs_reports_retrieval_and_json_errors():
+    def unreachable(url: str, timeout: float):
+        raise URLError("network down")
+
+    network_result = PackageDocsService(fetcher=unreachable).lookup("demo")
+    assert network_result.sources == []
+    assert network_result.note == "Unable to retrieve PyPI metadata: URLError."
+
+    def invalid_json(url: str, timeout: float):
+        return _Resp(b"not json")
+
+    json_result = PackageDocsService(fetcher=invalid_json).lookup("demo")
+    assert json_result.sources == []
+    assert json_result.note == "Unable to retrieve PyPI metadata: JSONDecodeError."