Merge pull request #5350 from pypa/issue-5349

Safer usages of pkg_resources

Author: oz123

news/5349.bugfix.rst

@@ -0,0 +1 @@
+Modernize ``pipenv`` path patch with ``importlib.util`` to eliminate import of ``pkg_resources``

pipenv/environment.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import importlib
+import importlib.util
 import itertools
 import json
 import operator
@@ -11,12 +12,11 @@
 from pathlib import Path
 from sysconfig import get_paths, get_python_version, get_scheme_names
 
-import pkg_resources
-
 import pipenv
 from pipenv.patched.pip._internal.commands.install import InstallCommand
 from pipenv.patched.pip._internal.index.package_finder import PackageFinder
 from pipenv.patched.pip._internal.req.req_uninstall import UninstallPathSet
+from pipenv.patched.pip._vendor import pkg_resources
 from pipenv.patched.pip._vendor.packaging.utils import canonicalize_name
 from pipenv.utils.constants import is_type_checking
 from pipenv.utils.indexes import prepare_pip_source_args
@@ -41,7 +41,6 @@
     from pipenv.project import Project, TPipfile, TSource
 
 BASE_WORKING_SET = pkg_resources.WorkingSet(sys.path)
-# TODO: Unittests for this class
 
 
 class Environment:
@@ -863,7 +862,7 @@ def run_activate_this(self):
                 exec(code, dict(__file__=activate_this))
 
     @contextlib.contextmanager
-    def activated(self, include_extras=True, extra_dists=None):
+    def activated(self):
         """Helper context manager to activate the environment.
 
         This context manager will set the following variables for the duration
@@ -882,16 +881,8 @@ def activated(self, include_extras=True, extra_dists=None):
         to `os.environ["PATH"]` to ensure that calls to `~Environment.run()` use the
         environment's path preferentially.
         """
-
-        if not extra_dists:
-            extra_dists = []
         original_path = sys.path
         original_prefix = sys.prefix
-        parent_path = Path(__file__).absolute().parent
-        vendor_dir = parent_path.joinpath("vendor").as_posix()
-        patched_dir = parent_path.joinpath("patched").as_posix()
-        parent_path = parent_path.as_posix()
-        self.add_dist("pip")
         prefix = self.prefix.as_posix()
         with vistir.contextmanagers.temp_environ(), vistir.contextmanagers.temp_path():
             os.environ["PATH"] = os.pathsep.join(
@@ -914,21 +905,6 @@ def activated(self, include_extras=True, extra_dists=None):
                     os.environ.pop("PYTHONHOME", None)
             sys.path = self.sys_path
             sys.prefix = self.sys_prefix
-            site.addsitedir(self.base_paths["purelib"])
-            pip = self.safe_import("pip")  # noqa
-            pip_vendor = self.safe_import("pip._vendor")
-            pep517_dir = os.path.join(os.path.dirname(pip_vendor.__file__), "pep517")
-            site.addsitedir(pep517_dir)
-            os.environ["PYTHONPATH"] = os.pathsep.join(
-                [os.environ.get("PYTHONPATH", self.base_paths["PYTHONPATH"]), pep517_dir]
-            )
-            if include_extras:
-                site.addsitedir(parent_path)
-                sys.path.extend([parent_path, patched_dir, vendor_dir])
-                extra_dists = list(self.extra_dists) + extra_dists
-                for extra_dist in extra_dists:
-                    if extra_dist not in self.get_working_set():
-                        extra_dist.activate(self.sys_path)
             try:
                 yield
             finally:

pipenv/patched/pip/_vendor/pkg_resources/LICENSE

@@ -61,7 +61,7 @@ def check(key, db, json, full_report, bare, stdin, files, cache, ignore, output,
     elif stdin:
         packages = list(read_requirements(sys.stdin))
     else:
-        import pkg_resources
+        import pipenv.patched.pip._vendor.pkg_resources as pkg_resources
         packages = [
             d for d in pkg_resources.working_set
             if d.key not in {"python", "wsgiref", "argparse"}
@@ -150,7 +150,7 @@ def license(key, db, json, bare, cache, files, proxyprotocol, proxyhost, proxypo
     if files:
         packages = list(itertools.chain.from_iterable(read_requirements(f, resolve=True) for f in files))
     else:
-        import pkg_resources
+        import pipenv.patched.pip._vendor.pkg_resources as pkg_resources
         packages = [
             d for d in pkg_resources.working_set
             if d.key not in {"python", "wsgiref", "argparse"}

pipenv/patched/safety/cli.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 import base64
@@ -23,6 +22,7 @@
 from pipenv.environment import Environment
 from pipenv.environments import Setting, is_in_virtualenv, normalize_pipfile_path
 from pipenv.patched.pip._internal.commands.install import InstallCommand
+from pipenv.patched.pip._vendor import pkg_resources
 from pipenv.utils.constants import is_type_checking
 from pipenv.utils.dependencies import (
     get_canonical_names,
@@ -37,6 +37,7 @@
     find_windows_executable,
     get_workon_home,
     is_virtual_environment,
+    load_path,
     looks_like_dir,
     safe_expandvars,
     system_which,
@@ -55,8 +56,6 @@
 if is_type_checking():
     from typing import Dict, List, Optional, Set, Text, Tuple, Union
 
-    import pkg_resources
-
     TSource = Dict[Text, Union[Text, bool]]
     TPackageEntry = Dict[str, Union[bool, str, List[str]]]
     TPackage = Dict[str, TPackageEntry]
@@ -243,11 +242,7 @@ def get_location_for_virtualenv(self) -> str:
 
     @property
     def working_set(self) -> pkg_resources.WorkingSet:
-        from pipenv.utils.shell import load_path
-
         sys_path = load_path(self.which("python"))
-        import pkg_resources
-
         return pkg_resources.WorkingSet(sys_path)
 
     @property

pipenv/project.py

@@ -1,3 +1,4 @@
+import importlib.util
 import json
 import logging
 import os
@@ -6,65 +7,13 @@
 os.environ["PIP_PYTHON_PATH"] = str(sys.executable)
 
 
-def find_site_path(pkg, site_dir=None):
-    import pkg_resources
-
-    if site_dir is None:
-        site_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-    working_set = pkg_resources.WorkingSet([site_dir] + sys.path[:])
-    for dist in working_set:
-        root = dist.location
-        base_name = dist.project_name if dist.project_name else dist.key
-        name = None
-        if "top_level.txt" in dist.metadata_listdir(""):
-            name = next(
-                iter(
-                    [
-                        line.strip()
-                        for line in dist.get_metadata_lines("top_level.txt")
-                        if line is not None
-                    ]
-                ),
-                None,
-            )
-        if name is None:
-            name = pkg_resources.safe_name(base_name).replace("-", "_")
-        if not any(pkg == _ for _ in [base_name, name]):
-            continue
-        path_options = [name, f"{name}.py"]
-        path_options = [os.path.join(root, p) for p in path_options if p is not None]
-        path = next(iter(p for p in path_options if os.path.exists(p)), None)
-        if path is not None:
-            return dist, path
-    return None, None
-
-
-def _patch_path(pipenv_site=None):
-    import site
-
-    pipenv_libdir = os.path.dirname(os.path.abspath(__file__))
-    pipenv_site_dir = os.path.dirname(pipenv_libdir)
-    if pipenv_site is not None:
-        pipenv_dist, pipenv_path = find_site_path("pipenv", site_dir=pipenv_site)
-    else:
-        pipenv_dist, pipenv_path = find_site_path("pipenv", site_dir=pipenv_site_dir)
-    if pipenv_dist is not None:
-        pipenv_dist.activate()
-    else:
-        site.addsitedir(
-            next(
-                iter(
-                    sitedir
-                    for sitedir in (pipenv_site, pipenv_site_dir)
-                    if sitedir is not None
-                ),
-                None,
-            )
-        )
-    if pipenv_path is not None:
-        pipenv_libdir = pipenv_path
-    for _dir in ("vendor", "patched", pipenv_libdir):
-        sys.path.insert(0, os.path.join(pipenv_libdir, _dir))
+def _ensure_modules():
+    spec = importlib.util.spec_from_file_location(
+        "pipenv", location=os.path.join(os.path.dirname(__file__), "__init__.py")
+    )
+    pipenv = importlib.util.module_from_spec(spec)
+    sys.modules["pipenv"] = pipenv
+    spec.loader.exec_module(pipenv)
 
 
 def get_parser():
@@ -838,7 +787,7 @@ def _main(
 def main(argv=None):
     parser = get_parser()
     parsed, remaining = parser.parse_known_args(argv)
-    _patch_path(pipenv_site=parsed.pipenv_site)
+    _ensure_modules()
     import warnings
 
     from pipenv.vendor.vistir.misc import replace_with_text_stream

pipenv/resolver.py

@@ -23,6 +23,7 @@
 from pipenv.patched.pip._internal.req.req_file import parse_requirements
 from pipenv.patched.pip._internal.utils.hashes import FAVORITE_HASH
 from pipenv.patched.pip._internal.utils.temp_dir import global_tempdir_manager
+from pipenv.patched.pip._vendor import pkg_resources
 from pipenv.project import Project
 from pipenv.vendor import click
 from pipenv.vendor.requirementslib import Requirement
@@ -1147,8 +1148,6 @@ def resolve_deps(
 
 @lru_cache()
 def get_pipenv_sitedir() -> Optional[str]:
-    import pkg_resources
-
     site_dir = next(
         iter(d for d in pkg_resources.working_set if d.key.lower() == "pipenv"), None
     )

pipenv/utils/resolver.py

@@ -10,7 +10,7 @@
 
 from __future__ import absolute_import
 
-from pkg_resources import get_distribution, DistributionNotFound
+from pipenv.patched.pip._vendor.pkg_resources import get_distribution, DistributionNotFound
 
 from pipenv.vendor.cerberus.validator import DocumentError, Validator
 from pipenv.vendor.cerberus.schema import rules_set_registry, schema_registry, SchemaError

pipenv/vendor/cerberus/__init__.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 from decimal import Decimal
-from pkg_resources import Distribution, DistributionNotFound
+from pipenv.patched.pip._vendor.pkg_resources import Distribution, DistributionNotFound
 
 from pytest import mark
 

pipenv/vendor/cerberus/tests/test_assorted.py

@@ -5,7 +5,7 @@
 from .models.pipfile import Pipfile
 from .models.requirements import Requirement
 
-__version__ = "2.0.1"
+__version__ = "2.0.2"
 
 
 logger = logging.getLogger(__name__)