File _service:tar_scm:backport-CVE-2024-6345.patch of Package python-setuptools

From 88807c7062788254f654ea8c03427adc859321f0 Mon Sep 17 00:00:00 2001
From: jaraco <jaraco@fosstodon.org>
Date: Tue, 30 Apr 2024 15:02:00 +0800
Subject: [PATCH] Modernize package_index VCS handling
https://github.com/pypa/setuptools/commit/88807c7062788254f654ea8c03427adc859321f0
https://github.com/pypa/setuptools/pull/4332

---
 changelog.d/4332.feature.rst          |   1 +
 setup.cfg                             |   1 +
 setuptools/package_index.py           | 146 ++++++++++++++------------
 setuptools/tests/test_packageindex.py |  56 +++++-----
 4 files changed, 108 insertions(+), 96 deletions(-)
 create mode 100644 changelog.d/4332.feature.rst

diff --git a/changelog.d/4332.feature.rst b/changelog.d/4332.feature.rst
new file mode 100644
index 0000000..1e612ec
--- /dev/null
+++ b/changelog.d/4332.feature.rst
@@ -0,0 +1 @@
+Modernized and refactored VCS handling in package_index.
diff --git a/setup.cfg b/setup.cfg
index c1d8a69..6787594 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -63,6 +63,7 @@ testing =
 	tomli-w>=1.0.0
 	pytest-timeout
 	pytest-perf
+	pytest-subprocess
 testing-integration = 
 	pytest
 	pytest-xdist
diff --git a/setuptools/package_index.py b/setuptools/package_index.py
index 3130ace..ae50db5 100644
--- a/setuptools/package_index.py
+++ b/setuptools/package_index.py
@@ -1,6 +1,7 @@
 """PyPI and direct package downloading."""
 
 import sys
+import subprocess
 import os
 import re
 import io
@@ -586,7 +587,7 @@ class PackageIndex(Environment):
             scheme = URL_SCHEME(spec)
             if scheme:
                 # It's a url, download it to tmpdir
-                found = self._download_url(scheme.group(1), spec, tmpdir)
+                found = self._download_url(spec, tmpdir)
                 base, fragment = egg_info_for_url(spec)
                 if base.endswith('.py'):
                     found = self.gen_setup(found, fragment, tmpdir)
@@ -813,7 +814,7 @@ class PackageIndex(Environment):
             else:
                 raise DistutilsError("Download error for %s: %s" % (url, v)) from v
 
-    def _download_url(self, scheme, url, tmpdir):
+    def _download_url(self, url, tmpdir):
         # Determine download filename
         #
         name, fragment = egg_info_for_url(url)
@@ -828,19 +829,60 @@ class PackageIndex(Environment):
 
         filename = os.path.join(tmpdir, name)
 
-        # Download the file
-        #
-        if scheme == 'svn' or scheme.startswith('svn+'):
-            return self._download_svn(url, filename)
-        elif scheme == 'git' or scheme.startswith('git+'):
-            return self._download_git(url, filename)
-        elif scheme.startswith('hg+'):
-            return self._download_hg(url, filename)
-        elif scheme == 'file':
-            return urllib.request.url2pathname(urllib.parse.urlparse(url)[2])
-        else:
-            self.url_ok(url, True)  # raises error if not allowed
-            return self._attempt_download(url, filename)
+        return self._download_vcs(url, filename) or self._download_other(url, filename)
+
+
+    @staticmethod
+    def _resolve_vcs(url):
+        """
+        >>> rvcs = PackageIndex._resolve_vcs
+        >>> rvcs('git+http://foo/bar')
+        'git'
+        >>> rvcs('hg+https://foo/bar')
+        'hg'
+        >>> rvcs('git:myhost')
+        'git'
+        >>> rvcs('hg:myhost')
+        >>> rvcs('http://foo/bar')
+        """
+        scheme = urllib.parse.urlsplit(url).scheme
+        pre, sep, post = scheme.partition('+')
+        # svn and git have their own protocol; hg does not
+        allowed = set(['svn', 'git'] + ['hg'] * bool(sep))
+        return next(iter({pre} & allowed), None)
+
+    def _download_vcs(self, url, spec_filename):
+        vcs = self._resolve_vcs(url)
+        if not vcs:
+            return
+        if vcs == 'svn':
+            raise DistutilsError(
+                f"Invalid config, SVN download is not supported: {url}"
+            )
+
+        filename, _, _ = spec_filename.partition('#')
+        url, rev = self._vcs_split_rev_from_url(url)
+
+        self.info(f"Doing {vcs} clone from {url} to {filename}")
+        subprocess.check_call([vcs, 'clone', '--quiet', url, filename])
+
+        co_commands = dict(
+            git=[vcs, '-C', filename, 'checkout', '--quiet', rev],
+            hg=[vcs, '--cwd', filename, 'up', '-C', '-r', rev, '-q'],
+        )
+        if rev is not None:
+            self.info(f"Checking out {rev}")
+            subprocess.check_call(co_commands[vcs])
+
+        return filename
+
+    def _download_other(self, url, filename):
+        scheme = urllib.parse.urlsplit(url).scheme
+        if scheme == 'file':  # pragma: no cover
+            return urllib.request.url2pathname(urllib.parse.urlparse(url).path)
+        # raise error if not allowed
+        self.url_ok(url, True)
+        return self._attempt_download(url, filename)
 
     def scan_url(self, url):
         self.process_url(url, True)
@@ -856,64 +898,36 @@ class PackageIndex(Environment):
         os.unlink(filename)
         raise DistutilsError(f"Unexpected HTML page found at {url}")
 
-    def _download_svn(self, url, _filename):
-        raise DistutilsError(f"Invalid config, SVN download is not supported: {url}")
-
     @staticmethod
-    def _vcs_split_rev_from_url(url, pop_prefix=False):
-        scheme, netloc, path, query, frag = urllib.parse.urlsplit(url)
+    def _vcs_split_rev_from_url(url):
+        """
+        Given a possible VCS URL, return a clean URL and resolved revision if any.
+        >>> vsrfu = PackageIndex._vcs_split_rev_from_url
+        >>> vsrfu('git+https://github.com/pypa/setuptools@v69.0.0#egg-info=setuptools')
+        ('https://github.com/pypa/setuptools', 'v69.0.0')
+        >>> vsrfu('git+https://github.com/pypa/setuptools#egg-info=setuptools')
+        ('https://github.com/pypa/setuptools', None)
+        >>> vsrfu('http://foo/bar')
+        ('http://foo/bar', None)
+        """
+        parts = urllib.parse.urlsplit(url)
 
-        scheme = scheme.split('+', 1)[-1]
+        clean_scheme = parts.scheme.split('+', 1)[-1]
 
         # Some fragment identification fails
-        path = path.split('#', 1)[0]
-
-        rev = None
-        if '@' in path:
-            path, rev = path.rsplit('@', 1)
-
-        # Also, discard fragment
-        url = urllib.parse.urlunsplit((scheme, netloc, path, query, ''))
+        no_fragment_path, _, _ = parts.path.partition('#')
 
-        return url, rev
+        pre, sep, post = no_fragment_path.rpartition('@')
+        clean_path, rev = (pre, post) if sep else (post, None)
 
-    def _download_git(self, url, filename):
-        filename = filename.split('#', 1)[0]
-        url, rev = self._vcs_split_rev_from_url(url, pop_prefix=True)
-
-        self.info("Doing git clone from %s to %s", url, filename)
-        os.system("git clone --quiet %s %s" % (url, filename))
-
-        if rev is not None:
-            self.info("Checking out %s", rev)
-            os.system(
-                "git -C %s checkout --quiet %s"
-                % (
-                    filename,
-                    rev,
-                )
-            )
+        resolved = parts._replace(
+            scheme=clean_scheme,
+            path=clean_path,
+            # discard the fragment
+            fragment='',
+        ).geturl()
 
-        return filename
-
-    def _download_hg(self, url, filename):
-        filename = filename.split('#', 1)[0]
-        url, rev = self._vcs_split_rev_from_url(url, pop_prefix=True)
-
-        self.info("Doing hg clone from %s to %s", url, filename)
-        os.system("hg clone --quiet %s %s" % (url, filename))
-
-        if rev is not None:
-            self.info("Updating to %s", rev)
-            os.system(
-                "hg --cwd %s up -C -r %s -q"
-                % (
-                    filename,
-                    rev,
-                )
-            )
-
-        return filename
+        return resolved, rev
 
     def debug(self, msg, *args):
         log.debug(msg, *args)
diff --git a/setuptools/tests/test_packageindex.py b/setuptools/tests/test_packageindex.py
index f1fa745..a7d2b5d 100644
--- a/setuptools/tests/test_packageindex.py
+++ b/setuptools/tests/test_packageindex.py
@@ -5,7 +5,6 @@ import platform
 import urllib.request
 import urllib.error
 import http.client
-from unittest import mock
 
 import pytest
 
@@ -186,49 +185,46 @@ class TestPackageIndex:
             assert dists[0].version == ''
             assert dists[1].version == vc
 
-    def test_download_git_with_rev(self, tmpdir):
+    def test_download_git_with_rev(self, tmp_path, fp):
         url = 'git+https://github.example/group/project@master#egg=foo'
         index = setuptools.package_index.PackageIndex()
 
-        with mock.patch("os.system") as os_system_mock:
-            result = index.download(url, str(tmpdir))
+        expected_dir = tmp_path / 'project@master'
+        fp.register([
+            'git',
+            'clone',
+            '--quiet',
+            'https://github.example/group/project',
+            expected_dir,
+        ])
+        fp.register(['git', '-C', expected_dir, 'checkout', '--quiet', 'master'])
 
-        os_system_mock.assert_called()
+        result = index.download(url, tmp_path)
 
-        expected_dir = str(tmpdir / 'project@master')
-        expected = (
-            'git clone --quiet ' 'https://github.example/group/project {expected_dir}'
-        ).format(**locals())
-        first_call_args = os_system_mock.call_args_list[0][0]
-        assert first_call_args == (expected,)
+        assert result == str(expected_dir)
+        assert len(fp.calls) == 2
 
-        tmpl = 'git -C {expected_dir} checkout --quiet master'
-        expected = tmpl.format(**locals())
-        assert os_system_mock.call_args_list[1][0] == (expected,)
-        assert result == expected_dir
-
-    def test_download_git_no_rev(self, tmpdir):
+    def test_download_git_no_rev(self, tmp_path, fp):
         url = 'git+https://github.example/group/project#egg=foo'
         index = setuptools.package_index.PackageIndex()
 
-        with mock.patch("os.system") as os_system_mock:
-            result = index.download(url, str(tmpdir))
-
-        os_system_mock.assert_called()
-
-        expected_dir = str(tmpdir / 'project')
-        expected = (
-            'git clone --quiet ' 'https://github.example/group/project {expected_dir}'
-        ).format(**locals())
-        os_system_mock.assert_called_once_with(expected)
-
-    def test_download_svn(self, tmpdir):
+        expected_dir = tmp_path / 'project'
+        fp.register([
+            'git',
+            'clone',
+            '--quiet',
+            'https://github.example/group/project',
+            expected_dir,
+        ])
+        index.download(url, tmp_path)
+
+    def test_download_svn(self, tmp_path):
         url = 'svn+https://svn.example/project#egg=foo'
         index = setuptools.package_index.PackageIndex()
 
         msg = r".*SVN download is not supported.*"
         with pytest.raises(distutils.errors.DistutilsError, match=msg):
-            index.download(url, str(tmpdir))
+            index.download(url, tmp_path)
 
 
 class TestContentCheckers:
-- 
2.33.0