CVE-2026-6357

@@ -0,0 +1,2 @@
+Split the pip self-version check into a fetch phase before the command
+body and an emit phase afterwards.

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import logging.config
 import optparse
 import os
 import sys
 import traceback
+from collections.abc import Iterator
 from optparse import Values
 from typing import Callable
 
@@ -80,24 +82,24 @@ def __init__(self, name: str, summary: str, isolated: bool = False) -> None:
     def add_options(self) -> None:
         pass
 
-    def handle_pip_version_check(self, options: Values) -> None:
+    @contextlib.contextmanager
+    def pip_version_check(self, options: Values, args: list[str]) -> Iterator[None]:
         """
         This is a no-op so that commands by default do not do the pip version
         check.
         """
         # Make sure we do the pip version check if the index_group options
         # are present.
         assert not hasattr(options, "no_index")
+        yield
 
     def run(self, options: Values, args: list[str]) -> int:
         raise NotImplementedError
 
     def _run_wrapper(self, level_number: int, options: Values, args: list[str]) -> int:
         def _inner_run() -> int:
-            try:
+            with self.pip_version_check(options, args):
                 return self.run(options, args)
-            finally:
-                self.handle_pip_version_check(options)
 
         if options.debug_mode:
             rich_traceback.install(show_locals=True)

@@ -8,8 +8,10 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import os
+from collections.abc import Iterator
 from functools import lru_cache
 from optparse import Values
 from typing import TYPE_CHECKING
@@ -25,6 +27,7 @@
     from pip._vendor.packaging.utils import NormalizedName
 
     from pip._internal.network.session import PipSession
+    from pip._internal.self_outdated_check import UpgradePrompt
 
 logger = logging.getLogger(__name__)
 
@@ -134,10 +137,18 @@ def _build_session(
         return session
 
 
-def _pip_self_version_check(session: PipSession, options: Values) -> None:
-    from pip._internal.self_outdated_check import pip_self_version_check as check
+def _pip_self_version_check_fetch(
+    session: PipSession, options: Values
+) -> UpgradePrompt | None:
+    from pip._internal.self_outdated_check import pip_self_version_check_fetch
 
-    check(session, options)
+    return pip_self_version_check_fetch(session, options)
+
+
+def _pip_self_version_check_emit(upgrade_prompt: UpgradePrompt | None) -> None:
+    from pip._internal.self_outdated_check import pip_self_version_check_emit
+
+    pip_self_version_check_emit(upgrade_prompt)
 
 
 class IndexGroupCommand(Command, SessionCommandMixin):
@@ -164,7 +175,8 @@ def should_exclude_prerelease(
         # No specific setting: exclude prereleases by default
         return True
 
-    def handle_pip_version_check(self, options: Values) -> None:
+    @contextlib.contextmanager
+    def pip_version_check(self, options: Values, args: list[str]) -> Iterator[None]:
         """
         Do the pip version check if not disabled.
 
@@ -174,17 +186,27 @@ def handle_pip_version_check(self, options: Values) -> None:
         assert hasattr(options, "no_index")
 
         if options.disable_pip_version_check or options.no_index:
+            yield
             return
 
+        upgrade_prompt: UpgradePrompt | None = None
         try:
-            # Otherwise, check if we're using the latest version of pip available.
             session = self._build_session(
                 options,
                 retries=0,
                 timeout=min(5, options.timeout),
             )
             with session:
-                _pip_self_version_check(session, options)
+                upgrade_prompt = _pip_self_version_check_fetch(session, options)
         except Exception:
             logger.warning("There was an error checking the latest version of pip.")
             logger.debug("See below for error", exc_info=True)
+
+        try:
+            yield
+        finally:
+            try:
+                _pip_self_version_check_emit(upgrade_prompt)
+            except Exception:
+                logger.warning("There was an error checking the latest version of pip.")
+                logger.debug("See below for error", exc_info=True)

@@ -1,14 +1,17 @@
 from __future__ import annotations
 
+import contextlib
 import errno
 import json
 import operator
 import os
 import shutil
 import site
+from collections.abc import Iterator
 from optparse import SUPPRESS_HELP, Values
 from pathlib import Path
 
+from pip._vendor.packaging.requirements import InvalidRequirement, Requirement
 from pip._vendor.packaging.utils import canonicalize_name
 from pip._vendor.requests.exceptions import InvalidProxyURL
 from pip._vendor.rich import print_json
@@ -63,6 +66,14 @@
 logger = getLogger(__name__)
 
 
+def _arg_refers_to_pip(arg: str) -> bool:
+    try:
+        req = Requirement(arg)
+    except InvalidRequirement:
+        return False
+    return canonicalize_name(req.name) == "pip"
+
+
 class InstallCommand(RequirementCommand):
     """
     Install packages from:
@@ -278,6 +289,17 @@ def add_options(self) -> None:
             ),
         )
 
+    @contextlib.contextmanager
+    def pip_version_check(self, options: Values, args: list[str]) -> Iterator[None]:
+        # Skip the self-version check when pip itself is a requirement. The
+        # running pip may be replaced mid-command, and the upgrade prompt
+        # is redundant.
+        if any(_arg_refers_to_pip(arg) for arg in args):
+            yield
+            return
+        with super().pip_version_check(options, args):
+            yield
+
     @with_cleanup
     def run(self, options: Values, args: list[str]) -> int:
         if options.use_user_site and options.target_dir is not None:

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
-from collections.abc import Generator, Sequence
+from collections.abc import Generator, Iterator, Sequence
 from email.parser import Parser
 from optparse import Values
 from typing import TYPE_CHECKING, cast
@@ -135,9 +136,13 @@ def add_options(self) -> None:
         self.parser.insert_option_group(0, selection_opts)
         self.parser.insert_option_group(0, self.cmd_opts)
 
-    def handle_pip_version_check(self, options: Values) -> None:
-        if options.outdated or options.uptodate:
-            super().handle_pip_version_check(options)
+    @contextlib.contextmanager
+    def pip_version_check(self, options: Values, args: list[str]) -> Iterator[None]:
+        if not (options.outdated or options.uptodate):
+            yield
+            return
+        with super().pip_version_check(options, args):
+            yield
 
     def _build_package_finder(
         self, options: Values, session: PipSession

@@ -1,15 +1,13 @@
 from __future__ import annotations
 
 import datetime
-import functools
 import hashlib
 import json
 import logging
 import optparse
 import os.path
 import sys
 from dataclasses import dataclass
-from typing import Callable
 
 from pip._vendor.packaging.version import Version
 from pip._vendor.packaging.version import parse as parse_version
@@ -156,16 +154,6 @@ def __rich__(self) -> Group:
         )
 
 
-def was_installed_by_pip(pkg: str) -> bool:
-    """Checks whether pkg was installed by pip
-
-    This is used not to display the upgrade message when pip is in fact
-    installed by system package manager, such as dnf on Fedora.
-    """
-    dist = get_default_environment().get_distribution(pkg)
-    return dist is not None and "pip" == dist.installer
-
-
 def _get_current_remote_pip_version(
     session: PipSession, options: optparse.Values
 ) -> str | None:
@@ -194,28 +182,15 @@ def _get_current_remote_pip_version(
     return str(best_candidate.version)
 
 
-def _self_version_check_logic(
-    *,
-    state: SelfCheckState,
-    current_time: datetime.datetime,
-    local_version: Version,
-    get_remote_version: Callable[[], str | None],
+def _compute_upgrade_prompt(
+    local_version: Version, remote_version_str: str, installed_by_pip: bool
 ) -> UpgradePrompt | None:
-    remote_version_str = state.get(current_time)
-    if remote_version_str is None:
-        remote_version_str = get_remote_version()
-        if remote_version_str is None:
-            logger.debug("No remote pip version found")
-            return None
-        state.set(remote_version_str, current_time)
-
     remote_version = parse_version(remote_version_str)
     logger.debug("Remote version of pip: %s", remote_version)
     logger.debug("Local version of pip:  %s", local_version)
+    logger.debug("Was pip installed by pip? %s", installed_by_pip)
 
-    pip_installed_by_pip = was_installed_by_pip("pip")
-    logger.debug("Was pip installed by pip? %s", pip_installed_by_pip)
-    if not pip_installed_by_pip:
+    if not installed_by_pip:
         return None  # Only suggest upgrade if pip is installed by pip.
 
     local_version_is_older = (
@@ -228,28 +203,44 @@ def _self_version_check_logic(
     return None
 
 
-def pip_self_version_check(session: PipSession, options: optparse.Values) -> None:
-    """Check for an update for pip.
+def pip_self_version_check_fetch(
+    session: PipSession, options: optparse.Values
+) -> UpgradePrompt | None:
+    """Compute the pip upgrade prompt, if any, before the command runs.
 
     Limit the frequency of checks to once per week. State is stored either in
     the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix
     of the pip script path.
+
+    Pair with :func:`pip_self_version_check_emit`, which displays the prompt
+    after the command body runs.
     """
     installed_dist = get_default_environment().get_distribution("pip")
     if not installed_dist:
-        return
+        return None
     try:
         check_externally_managed()
     except ExternallyManagedEnvironment:
-        return
+        return None
+
+    state = SelfCheckState(cache_dir=options.cache_dir)
+    current_time = datetime.datetime.now(datetime.timezone.utc)
+    remote_version_str = state.get(current_time)
+    if remote_version_str is None:
+        remote_version_str = _get_current_remote_pip_version(session, options)
+        if remote_version_str is None:
+            logger.debug("No remote pip version found")
+            return None
+        state.set(remote_version_str, current_time)
 
-    upgrade_prompt = _self_version_check_logic(
-        state=SelfCheckState(cache_dir=options.cache_dir),
-        current_time=datetime.datetime.now(datetime.timezone.utc),
+    return _compute_upgrade_prompt(
         local_version=installed_dist.version,
-        get_remote_version=functools.partial(
-            _get_current_remote_pip_version, session, options
-        ),
+        remote_version_str=remote_version_str,
+        installed_by_pip=installed_dist.installer == "pip",
     )
+
+
+def pip_self_version_check_emit(upgrade_prompt: UpgradePrompt | None) -> None:
+    """Emit the upgrade prompt captured by :func:`pip_self_version_check_fetch`."""
     if upgrade_prompt is not None:
         logger.warning("%s", upgrade_prompt, extra={"rich": True})

@@ -100,14 +100,14 @@ def test_raise_broken_stdout__debug_logging(
         assert "Traceback (most recent call last):" in stderr
 
 
-@patch("pip._internal.cli.index_command.Command.handle_pip_version_check")
-def test_handle_pip_version_check_called(mock_handle_version_check: Mock) -> None:
+@patch("pip._internal.cli.index_command.Command.pip_version_check")
+def test_pip_version_check_called(mock_version_check: Mock) -> None:
     """
-    Check that Command.handle_pip_version_check() is called.
+    Check that ``Command.pip_version_check()`` wraps the command body.
     """
     cmd = FakeCommand()
     cmd.main([])
-    mock_handle_version_check.assert_called_once()
+    mock_version_check.assert_called_once()
 
 
 def test_debug_enables_verbose_logs() -> None:

@@ -95,43 +95,69 @@ def has_option_no_index(command: Command) -> bool:
 @pytest.mark.parametrize(
     "disable_pip_version_check, no_index, expected_called",
     [
-        # pip_self_version_check() is only called when both
-        # disable_pip_version_check and no_index are False.
+        # The fetch phase only runs when both disable_pip_version_check
+        # and no_index are False.
         (False, False, True),
         (False, True, False),
         (True, False, False),
         (True, True, False),
     ],
 )
-@mock.patch("pip._internal.cli.index_command._pip_self_version_check")
-def test_index_group_handle_pip_version_check(
+@mock.patch("pip._internal.cli.index_command._pip_self_version_check_fetch")
+def test_index_group_pip_version_check(
     mock_version_check: mock.Mock,
     command_name: str,
     disable_pip_version_check: bool,
     no_index: bool,
     expected_called: bool,
 ) -> None:
     """
-    Test whether pip_self_version_check() is called when
-    handle_pip_version_check() is called, for each of the
-    IndexGroupCommand classes.
+    Test whether the pre-body fetch runs when ``pip_version_check()`` is
+    entered, for each of the IndexGroupCommand classes.
     """
     command = create_command(command_name)
     options = command.parser.get_default_values()
     options.disable_pip_version_check = disable_pip_version_check
     options.no_index = no_index
+    # Return None so the emit branch is a no-op.
+    mock_version_check.return_value = None
 
     # See test test_list_pip_version_check() below.
     if command_name == "list":
         expected_called = False
 
-    command.handle_pip_version_check(options)
+    with command.pip_version_check(options, []):
+        pass
     if expected_called:
         mock_version_check.assert_called_once()
     else:
         mock_version_check.assert_not_called()
 
 
+@mock.patch("pip._internal.cli.index_command._pip_self_version_check_fetch")
+def test_install_pip_version_check_skipped_when_pip_is_a_requirement(
+    mock_version_check: mock.Mock,
+) -> None:
+    """``pip install pip`` must skip the self-version check: the running pip
+    may be replaced before emit."""
+    command = create_command("install")
+    options = command.parser.get_default_values()
+    options.disable_pip_version_check = False
+    options.no_index = False
+
+    with command.pip_version_check(options, ["pip"]):
+        pass
+    mock_version_check.assert_not_called()
+
+    with command.pip_version_check(options, ["pip==25.0"]):
+        pass
+    mock_version_check.assert_not_called()
+
+    with command.pip_version_check(options, ["some-other-pkg"]):
+        pass
+    mock_version_check.assert_called_once()
+
+
 def test_requirement_commands() -> None:
     """
     Test which commands inherit from RequirementCommand.
@@ -144,7 +170,7 @@ def is_requirement_command(command: Command) -> bool:
 
 
 @pytest.mark.parametrize("flag", ["", "--outdated", "--uptodate"])
-@mock.patch("pip._internal.cli.index_command._pip_self_version_check")
+@mock.patch("pip._internal.cli.index_command._pip_self_version_check_fetch")
 @mock.patch.dict(os.environ, {"PIP_DISABLE_PIP_VERSION_CHECK": "no"})
 def test_list_pip_version_check(version_check_mock: mock.Mock, flag: str) -> None:
     """

@@ -7,18 +7,30 @@
 import sys
 from optparse import Values
 from pathlib import Path
-from unittest.mock import ANY, Mock, patch
+from unittest.mock import Mock, patch
 
 import pytest
 from freezegun import freeze_time
 
 from pip._vendor.packaging.version import Version
 
 from pip._internal import self_outdated_check
-from pip._internal.self_outdated_check import UpgradePrompt, pip_self_version_check
+from pip._internal.self_outdated_check import (
+    UpgradePrompt,
+    pip_self_version_check_emit,
+    pip_self_version_check_fetch,
+)
 from pip._internal.utils.misc import ExternallyManagedEnvironment
 
 
+def _make_installed_dist(version: str, installer: str = "pip") -> Mock:
+    """Build a stand-in for the installed pip distribution."""
+    installed_dist = Mock()
+    installed_dist.version = Version(version)
+    installed_dist.installer = installer
+    return installed_dist
+
+
 @pytest.mark.parametrize(
     "key, expected",
     [
@@ -37,32 +49,60 @@ def test_get_statefile_name_known_values(key: str, expected: str) -> None:
 
 
 @freeze_time("1970-01-02T11:00:00Z")
-@patch("pip._internal.self_outdated_check._self_version_check_logic")
+@patch("pip._internal.self_outdated_check._get_current_remote_pip_version")
 @patch("pip._internal.self_outdated_check.SelfCheckState")
+@patch("pip._internal.self_outdated_check.get_default_environment")
 @patch("pip._internal.self_outdated_check.check_externally_managed", new=lambda: None)
-def test_pip_self_version_check_calls_underlying_implementation(
-    mocked_state: Mock, mocked_function: Mock, tmpdir: Path
+def test_pip_self_version_check_fetch_calls_underlying_implementation(
+    mocked_env: Mock, mocked_state: Mock, mocked_get_remote: Mock, tmpdir: Path
 ) -> None:
     # GIVEN
     mock_session = Mock()
     fake_options = Values({"cache_dir": str(tmpdir)})
-    mocked_function.return_value = None
+    mocked_env.return_value.get_distribution.return_value = _make_installed_dist("1.0")
+    mocked_state.return_value.get.return_value = None
+    mocked_get_remote.return_value = "5.0"
 
     # WHEN
-    self_outdated_check.pip_self_version_check(mock_session, fake_options)
+    result = pip_self_version_check_fetch(mock_session, fake_options)
 
     # THEN
+    assert result == UpgradePrompt(old="1.0", new="5.0")
     mocked_state.assert_called_once_with(cache_dir=str(tmpdir))
-    mocked_function.assert_called_once_with(
-        state=mocked_state(cache_dir=str(tmpdir)),
-        current_time=datetime.datetime(
-            1970, 1, 2, 11, 0, 0, tzinfo=datetime.timezone.utc
-        ),
-        local_version=ANY,
-        get_remote_version=ANY,
+    mocked_get_remote.assert_called_once_with(mock_session, fake_options)
+    mocked_state.return_value.set.assert_called_once_with(
+        "5.0",
+        datetime.datetime(1970, 1, 2, 11, 0, 0, tzinfo=datetime.timezone.utc),
     )
 
 
+def test_pip_self_version_check_emit_logs_prompt(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # GIVEN
+    prompt = UpgradePrompt(old="1.0", new="2.0")
+
+    # WHEN
+    with caplog.at_level(logging.WARNING):
+        pip_self_version_check_emit(prompt)
+
+    # THEN
+    assert len(caplog.records) == 1
+    assert caplog.records[0].levelno == logging.WARNING
+
+
+def test_pip_self_version_check_emit_no_prompt_is_silent(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # WHEN
+    with caplog.at_level(logging.WARNING):
+        pip_self_version_check_emit(None)
+
+    # THEN
+    assert caplog.records == []
+
+
+@freeze_time("2000-01-01T00:00:00Z")
 @pytest.mark.parametrize(
     [  # noqa: PT006 - String representation is too long
         "installed_version",
@@ -94,37 +134,51 @@ def test_core_logic(
     should_show_prompt: bool,
     caplog: pytest.LogCaptureFixture,
     monkeypatch: pytest.MonkeyPatch,
+    tmpdir: Path,
 ) -> None:
     # GIVEN
+    installed_dist = _make_installed_dist(
+        installed_version, installer="pip" if installed_by_pip else "apt"
+    )
+    monkeypatch.setattr(
+        self_outdated_check,
+        "get_default_environment",
+        lambda: Mock(get_distribution=Mock(return_value=installed_dist)),
+    )
+    monkeypatch.setattr(self_outdated_check, "check_externally_managed", lambda: None)
     monkeypatch.setattr(
-        self_outdated_check, "was_installed_by_pip", lambda _: installed_by_pip
+        self_outdated_check,
+        "_get_current_remote_pip_version",
+        lambda session, options: remote_version,
     )
-    mock_state = Mock()
-    mock_state.get.return_value = stored_version
-    fake_time = datetime.datetime(2000, 1, 1, 0, 0, 0)
+    mock_state_instance = Mock()
+    mock_state_instance.get.return_value = stored_version
+    monkeypatch.setattr(
+        self_outdated_check,
+        "SelfCheckState",
+        Mock(return_value=mock_state_instance),
+    )
+    fake_time = datetime.datetime(2000, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
     version_that_should_be_checked = stored_version or remote_version
 
     # WHEN
     with caplog.at_level(logging.DEBUG):
-        return_value = self_outdated_check._self_version_check_logic(
-            state=mock_state,
-            current_time=fake_time,
-            local_version=Version(installed_version),
-            get_remote_version=lambda: remote_version,
+        return_value = pip_self_version_check_fetch(
+            session=Mock(), options=Values({"cache_dir": str(tmpdir)})
         )
 
     # THEN
-    mock_state.get.assert_called_once_with(fake_time)
+    mock_state_instance.get.assert_called_once_with(fake_time)
     assert caplog.messages == [
         f"Remote version of pip: {version_that_should_be_checked}",
         f"Local version of pip:  {installed_version}",
         f"Was pip installed by pip? {installed_by_pip}",
     ]
 
     if stored_version:
-        mock_state.set.assert_not_called()
+        mock_state_instance.set.assert_not_called()
     else:
-        mock_state.set.assert_called_once_with(
+        mock_state_instance.set.assert_called_once_with(
             version_that_should_be_checked, fake_time
         )
 
@@ -196,13 +250,29 @@ def test_writes_expected_statefile(self, tmpdir: Path) -> None:
         assert statefile_permissions == selfcheckdir_permissions == cache_permissions
 
 
-@patch("pip._internal.self_outdated_check._self_version_check_logic")
-def test_suppressed_by_externally_managed(mocked_function: Mock, tmpdir: Path) -> None:
-    mocked_function.return_value = UpgradePrompt(old="1.0", new="2.0")
+@patch("pip._internal.self_outdated_check._get_current_remote_pip_version")
+@patch("pip._internal.self_outdated_check.get_default_environment")
+def test_fetch_suppressed_by_externally_managed(
+    mocked_env: Mock, mocked_get_remote: Mock, tmpdir: Path
+) -> None:
+    mocked_env.return_value.get_distribution.return_value = _make_installed_dist("1.0")
     fake_options = Values({"cache_dir": str(tmpdir)})
     with patch(
         "pip._internal.self_outdated_check.check_externally_managed",
         side_effect=ExternallyManagedEnvironment("nope"),
     ):
-        pip_self_version_check(session=Mock(), options=fake_options)
-    mocked_function.assert_not_called()
+        result = pip_self_version_check_fetch(session=Mock(), options=fake_options)
+    assert result is None
+    mocked_get_remote.assert_not_called()
+
+
+@patch("pip._internal.self_outdated_check._get_current_remote_pip_version")
+@patch("pip._internal.self_outdated_check.get_default_environment")
+def test_fetch_skipped_when_pip_not_installed(
+    mocked_env: Mock, mocked_get_remote: Mock, tmpdir: Path
+) -> None:
+    mocked_env.return_value.get_distribution.return_value = None
+    fake_options = Values({"cache_dir": str(tmpdir)})
+    result = pip_self_version_check_fetch(session=Mock(), options=fake_options)
+    assert result is None
+    mocked_get_remote.assert_not_called()