0.3.3: treat the package index as untrusted in mpat diff

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.3.3
+
+- The archive download behind `mpat diff` and `mpat check --diff` treats the
+  package index as untrusted input. It only picks archives whose name is the
+  locked project's, so a filename like `../../x-1.0.tar.gz` can no longer write
+  outside `MPAT_CACHE_DIR`. It only follows `http` and `https` links, so an
+  index cannot point it at a local `file://` path. Error messages drop query
+  strings and credentials from URLs, so a signed download URL does not end up
+  in a CI log.
+
 ## 0.3.2
 
 - `mpat check` lists only the targets that are not `ok` and ends with an

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mpat"
-version = "0.3.2"
+version = "0.3.3"
 description = "Lockfile-based upstream drift tracking for Python monkeypatches"
 readme = "README.md"
 license = "MIT"

src/mpat/__init__.py

@@ -15,7 +15,7 @@
 )
 from mpat._until import Probe, Until, Version
 
-__version__ = "0.3.2"
+__version__ = "0.3.3"
 
 __all__ = [
     "AlreadyPatched",

src/mpat/_upstream.py

@@ -59,6 +59,7 @@
 _INIT = "__init__.py"
 _PY = ".py"
 _AUTHORIZATION = "Authorization"
+_SCHEMES = ("http", "https")
 
 
 class UpstreamUnavailable(MpatError):
@@ -154,9 +155,18 @@ def redirect_request(
         return new
 
 
+def _redacted(url: str) -> str:
+    """Signed download URLs carry their token in the query string; keep it out of CI logs."""
+    parts = urllib.parse.urlsplit(url)
+    netloc = parts.netloc.rpartition("@")[2]
+    return urllib.parse.urlunsplit((parts.scheme, netloc, parts.path, "", ""))
+
+
 def _get(
     url: str, *, auth: str | None, auth_host: str | None, accept: str | None = None
 ) -> tuple[bytes, str]:
+    if urllib.parse.urlsplit(url).scheme not in _SCHEMES:
+        raise UpstreamUnavailable(f"refusing to download {_redacted(url)}: only http and https")
     request = urllib.request.Request(url)
     if accept:
         request.add_header("Accept", accept)
@@ -167,7 +177,7 @@ def _get(
         with opener.open(request, timeout=TIMEOUT_SECONDS) as response:
             return response.read(), response.headers.get_content_type()
     except (urllib.error.URLError, http.client.HTTPException, OSError, ValueError) as exc:
-        raise UpstreamUnavailable(f"cannot fetch {url}: {exc}") from exc
+        raise UpstreamUnavailable(f"cannot fetch {_redacted(url)}: {exc}") from exc
 
 
 def _links(body: bytes, content_type: str, *, page: str) -> list[_Link]:
@@ -190,25 +200,35 @@ def _links(body: bytes, content_type: str, *, page: str) -> list[_Link]:
     ]
 
 
-def _archive_version(filename: str) -> tuple[Version, bool, bool] | None:
-    """(version, is_wheel, is_pure) for a distribution filename, None for anything else."""
+def _archive_version(filename: str, *, dist: str) -> tuple[Version, bool, bool] | None:
+    """(version, is_wheel, is_pure) for an archive of `dist`, None for anything else.
+
+    The filename comes from the index and becomes a path in the cache, so it must
+    be a bare filename of this very project.
+    """
+    if Path(filename).name != filename or "\\" in filename:
+        return None
     try:
         if filename.endswith(".whl"):
-            _, version, _, tags = parse_wheel_filename(filename)
-            return version, True, any(str(tag) == _PURE_WHEEL_TAG for tag in tags)
-        if filename.endswith(_SDIST_SUFFIXES):
-            _, version = parse_sdist_filename(filename)
-            return version, False, False
+            name, version, _, tags = parse_wheel_filename(filename)
+            is_wheel, is_pure = True, any(str(tag) == _PURE_WHEEL_TAG for tag in tags)
+        elif filename.endswith(_SDIST_SUFFIXES):
+            name, version = parse_sdist_filename(filename)
+            is_wheel, is_pure = False, False
+        else:
+            return None
     except (InvalidWheelFilename, InvalidSdistFilename, InvalidVersion):
         return None
-    return None
+    if name != canonicalize_name(dist):
+        return None
+    return version, is_wheel, is_pure
 
 
-def _pick(links: list[_Link], version: Version) -> _Link | None:
+def _pick(links: list[_Link], *, dist: str, version: Version) -> _Link | None:
     """Prefer a pure wheel, then any wheel, then an sdist, of exactly `version`."""
     ranked: list[tuple[bool, bool, str, _Link]] = []
     for link in links:
-        parsed = _archive_version(link.filename)
+        parsed = _archive_version(link.filename, dist=dist)
         if parsed is None:
             continue
         found, is_wheel, is_pure = parsed
@@ -227,7 +247,7 @@ def _archive(dist: str, version: str) -> Path:
     auth = auth or (_netrc_auth(index_host) if index_host else None)
     page = f"{index.rstrip('/')}/{canonicalize_name(dist)}/"
     body, content_type = _get(page, auth=auth, auth_host=index_host, accept=_ACCEPT)
-    link = _pick(_links(body, content_type, page=page), wanted)
+    link = _pick(_links(body, content_type, page=page), dist=dist, version=wanted)
     if link is None:
         raise UpstreamUnavailable(f"no archive for {dist} {version} on {index}")
     cached = _cache_dir() / link.filename

tests/test_upstream.py

@@ -136,3 +136,43 @@ def test_unwritable_cache_is_unavailable(upstream, index, locked, monkeypatch, t
     upstream.build_wheel(index.files, "1.0")
     with pytest.raises(_upstream.UpstreamUnavailable, match="cannot cache"):
         _upstream.locked_source(locked)
+
+
+def test_archive_of_another_project_is_ignored(upstream, index, locked):
+    upstream.build_wheel(index.files, "1.0")
+    (index.files / "fakeup-1.0-py3-none-any.whl").rename(index.files / "other-1.0.tar.gz")
+    with pytest.raises(_upstream.UpstreamUnavailable, match="no archive for fakeup 1.0"):
+        _upstream.locked_source(locked)
+
+
+def test_index_filenames_cannot_escape_the_cache(upstream, index, locked, monkeypatch):
+    links = [
+        _upstream._Link(filename="../../fakeup-1.0.tar.gz", url=f"{index.url}/x"),
+        _upstream._Link(filename="fakeup/../../fakeup-1.0.tar.gz", url=f"{index.url}/x"),
+    ]
+    monkeypatch.setattr(_upstream, "_links", lambda *a, **k: links)
+    with pytest.raises(_upstream.UpstreamUnavailable, match="no archive for fakeup 1.0"):
+        _upstream.locked_source(locked)
+
+
+def test_non_http_download_links_are_refused(upstream, index, locked, monkeypatch, tmp_path):
+    secret = tmp_path / "secret"
+    secret.write_text("x")
+    links = [_upstream._Link(filename="fakeup-1.0.tar.gz", url=secret.as_uri())]
+    monkeypatch.setattr(_upstream, "_links", lambda *a, **k: links)
+    with pytest.raises(_upstream.UpstreamUnavailable, match="refusing to download"):
+        _upstream.locked_source(locked)
+
+
+def test_errors_do_not_print_signed_query_strings(upstream, index, locked, monkeypatch):
+    links = [
+        _upstream._Link(
+            filename="fakeup-1.0.tar.gz",
+            url=f"{index.url.removesuffix('/simple')}/missing?X-Amz-Signature=s3cret",
+        )
+    ]
+    monkeypatch.setattr(_upstream, "_links", lambda *a, **k: links)
+    with pytest.raises(_upstream.UpstreamUnavailable) as raised:
+        _upstream.locked_source(locked)
+    assert "s3cret" not in str(raised.value)
+    assert "/missing" in str(raised.value)

tests/test_version.py

@@ -6,4 +6,4 @@
 
 def test_version_matches_pyproject():
     data = tomllib.loads((Path(__file__).parents[1] / "pyproject.toml").read_text())
-    assert data["project"]["version"] == mpat.__version__ == "0.3.2"
+    assert data["project"]["version"] == mpat.__version__ == "0.3.3"