Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py """Contains purely network-related utilities."""	49 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 auth.py """Network Authentication Helpers Contains interface (MultiDomainBasicAuth) and associated glue code for providing credentials in the context of network requests. """ from __future__ import annotations import logging import os import shutil import subprocess import sysconfig import typing import urllib.parse from abc import ABC, abstractmethod from functools import cache from os.path import commonprefix from pathlib import Path from typing import Any, NamedTuple from pip._vendor.requests.auth import AuthBase, HTTPBasicAuth from pip._vendor.requests.models import Request, Response from pip._vendor.requests.utils import get_netrc_auth from pip._internal.utils.logging import getLogger from pip._internal.utils.misc import ( ask, ask_input, ask_password, remove_auth_from_url, split_auth_netloc_from_url, ) from pip._internal.vcs.versioncontrol import AuthInfo logger = getLogger(__name__) KEYRING_DISABLED = False class Credentials(NamedTuple): url: str username: str password: str class KeyRingBaseProvider(ABC): """Keyring base provider interface""" has_keyring: bool @abstractmethod def get_auth_info(self, url: str, username: str \| None) -> AuthInfo \| None: ... @abstractmethod def save_auth_info(self, url: str, username: str, password: str) -> None: ... class KeyRingNullProvider(KeyRingBaseProvider): """Keyring null provider""" has_keyring = False def get_auth_info(self, url: str, username: str \| None) -> AuthInfo \| None: return None def save_auth_info(self, url: str, username: str, password: str) -> None: return None class KeyRingPythonProvider(KeyRingBaseProvider): """Keyring interface which uses locally imported `keyring`""" has_keyring = True def __init__(self) -> None: import keyring self.keyring = keyring def get_auth_info(self, url: str, username: str \| None) -> AuthInfo \| None: # Support keyring's get_credential interface which supports getting # credentials without a username. This is only available for # keyring>=15.2.0. if hasattr(self.keyring, "get_credential"): logger.debug("Getting credentials from keyring for %s", url) cred = self.keyring.get_credential(url, username) if cred is not None: return cred.username, cred.password return None if username is not None: logger.debug("Getting password from keyring for %s", url) password = self.keyring.get_password(url, username) if password: return username, password return None def save_auth_info(self, url: str, username: str, password: str) -> None: self.keyring.set_password(url, username, password) class KeyRingCliProvider(KeyRingBaseProvider): """Provider which uses `keyring` cli Instead of calling the keyring package installed alongside pip we call keyring on the command line which will enable pip to use which ever installation of keyring is available first in PATH. """ has_keyring = True def __init__(self, cmd: str) -> None: self.keyring = cmd def get_auth_info(self, url: str, username: str \| None) -> AuthInfo \| None: # This is the default implementation of keyring.get_credential # https://github.com/jaraco/keyring/blob/97689324abcf01bd1793d49063e7ca01e03d7d07/keyring/backend.py#L134-L139 if username is not None: password = self._get_password(url, username) if password is not None: return username, password return None def save_auth_info(self, url: str, username: str, password: str) -> None: return self._set_password(url, username, password) def _get_password(self, service_name: str, username: str) -> str \| None: """Mirror the implementation of keyring.get_password using cli""" if self.keyring is None: return None cmd = [self.keyring, "get", service_name, username] env = os.environ.copy() env["PYTHONIOENCODING"] = "utf-8" res = subprocess.run( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, env=env, ) if res.returncode: return None return res.stdout.decode("utf-8").strip(os.linesep) def _set_password(self, service_name: str, username: str, password: str) -> None: """Mirror the implementation of keyring.set_password using cli""" if self.keyring is None: return None env = os.environ.copy() env["PYTHONIOENCODING"] = "utf-8" subprocess.run( [self.keyring, "set", service_name, username], input=f"{password}{os.linesep}".encode(), env=env, check=True, ) return None @cache def get_keyring_provider(provider: str) -> KeyRingBaseProvider: logger.verbose("Keyring provider requested: %s", provider) # keyring has previously failed and been disabled if KEYRING_DISABLED: provider = "disabled" if provider in ["import", "auto"]: try: impl = KeyRingPythonProvider() logger.verbose("Keyring provider set: import") return impl except ImportError: pass except Exception as exc: # In the event of an unexpected exception # we should warn the user msg = "Installed copy of keyring fails with exception %s" if provider == "auto": msg = msg + ", trying to find a keyring executable as a fallback" logger.warning(msg, exc, exc_info=logger.isEnabledFor(logging.DEBUG)) if provider in ["subprocess", "auto"]: cli = shutil.which("keyring") if cli and cli.startswith(sysconfig.get_path("scripts")): # all code within this function is stolen from shutil.which implementation @typing.no_type_check def PATH_as_shutil_which_determines_it() -> str: path = os.environ.get("PATH", None) if path is None: try: path = os.confstr("CS_PATH") except (AttributeError, ValueError): # os.confstr() or CS_PATH is not available path = os.defpath # bpo-35755: Don't use os.defpath if the PATH environment variable is # set to an empty string return path scripts = Path(sysconfig.get_path("scripts")) paths = [] for path in PATH_as_shutil_which_determines_it().split(os.pathsep): p = Path(path) try: if not p.samefile(scripts): paths.append(path) except FileNotFoundError: pass path = os.pathsep.join(paths) cli = shutil.which("keyring", path=path) if cli: logger.verbose("Keyring provider set: subprocess with executable %s", cli) return KeyRingCliProvider(cli) logger.verbose("Keyring provider set: disabled") return KeyRingNullProvider() class MultiDomainBasicAuth(AuthBase): def __init__( self, prompting: bool = True, index_urls: list[str] \| None = None, keyring_provider: str = "auto", ) -> None: self.prompting = prompting self.index_urls = index_urls self.keyring_provider = keyring_provider self.passwords: dict[str, AuthInfo] = {} # When the user is prompted to enter credentials and keyring is # available, we will offer to save them. If the user accepts, # this value is set to the credentials they entered. After the # request authenticates, the caller should call # ``save_credentials`` to save these. self._credentials_to_save: Credentials \| None = None @property def keyring_provider(self) -> KeyRingBaseProvider: return get_keyring_provider(self._keyring_provider) @keyring_provider.setter def keyring_provider(self, provider: str) -> None: # The free function get_keyring_provider has been decorated with # functools.cache. If an exception occurs in get_keyring_auth that # cache will be cleared and keyring disabled, take that into account # if you want to remove this indirection. self._keyring_provider = provider @property def use_keyring(self) -> bool: # We won't use keyring when --no-input is passed unless # a specific provider is requested because it might require # user interaction return self.prompting or self._keyring_provider not in ["auto", "disabled"] def _get_keyring_auth( self, url: str \| None, username: str \| None, ) -> AuthInfo \| None: """Return the tuple auth for a given url from keyring.""" # Do nothing if no url was provided if not url: return None try: return self.keyring_provider.get_auth_info(url, username) except Exception as exc: # Log the full exception (with stacktrace) at debug, so it'll only # show up when running in verbose mode. logger.debug("Keyring is skipped due to an exception", exc_info=True) # Always log a shortened version of the exception. logger.warning( "Keyring is skipped due to an exception: %s", str(exc), ) global KEYRING_DISABLED KEYRING_DISABLED = True get_keyring_provider.cache_clear() return None def _get_index_url(self, url: str) -> str \| None: """Return the original index URL matching the requested URL. Cached or dynamically generated credentials may work against the original index URL rather than just the netloc. The provided url should have had its username and password removed already. If the original index url had credentials then they will be included in the return value. Returns None if no matching index was found, or if --no-index was specified by the user. """ if not url or not self.index_urls: return None url = remove_auth_from_url(url).rstrip("/") + "/" parsed_url = urllib.parse.urlsplit(url) candidates = [] for index in self.index_urls: index = index.rstrip("/") + "/" parsed_index = urllib.parse.urlsplit(remove_auth_from_url(index)) if parsed_url == parsed_index: return index if parsed_url.netloc != parsed_index.netloc: continue candidate = urllib.parse.urlsplit(index) candidates.append(candidate) if not candidates: return None candidates.sort( reverse=True, key=lambda candidate: commonprefix( [ parsed_url.path, candidate.path, ] ).rfind("/"), ) return urllib.parse.urlunsplit(candidates[0]) def _get_new_credentials( self, original_url: str, , allow_netrc: bool = True, allow_keyring: bool = False, ) -> AuthInfo: """Find and return credentials for the specified URL.""" # Split the credentials and netloc from the url. url, netloc, url_user_password = split_auth_netloc_from_url( original_url, ) # Start with the credentials embedded in the url username, password = url_user_password if username is not None and password is not None: logger.debug("Found credentials in url for %s", netloc) return url_user_password # Find a matching index url for this request index_url = self._get_index_url(url) if index_url: # Split the credentials from the url. index_info = split_auth_netloc_from_url(index_url) if index_info: index_url, _, index_url_user_password = index_info logger.debug("Found index url %s", index_url) # If an index URL was found, try its embedded credentials if index_url and index_url_user_password[0] is not None: username, password = index_url_user_password if username is not None and password is not None: logger.debug("Found credentials in index url for %s", netloc) return index_url_user_password # Get creds from netrc if we still don't have them if allow_netrc: netrc_auth = get_netrc_auth(original_url) if netrc_auth: logger.debug("Found credentials in netrc for %s", netloc) return netrc_auth # If we don't have a password and keyring is available, use it. if allow_keyring: # The index url is more specific than the netloc, so try it first # fmt: off kr_auth = ( self._get_keyring_auth(index_url, username) or self._get_keyring_auth(netloc, username) ) # fmt: on if kr_auth: logger.debug("Found credentials in keyring for %s", netloc) return kr_auth return username, password def _get_url_and_credentials( self, original_url: str ) -> tuple[str, str \| None, str \| None]: """Return the credentials to use for the provided URL. If allowed, netrc and keyring may be used to obtain the correct credentials. Returns (url_without_credentials, username, password). Note that even if the original URL contains credentials, this function may return a different username and password. """ url, netloc, _ = split_auth_netloc_from_url(original_url) # Try to get credentials from original url username, password = self._get_new_credentials(original_url) # If credentials not found, use any stored credentials for this netloc. # Do this if either the username or the password is missing. # This accounts for the situation in which the user has specified # the username in the index url, but the password comes from keyring. if (username is None or password is None) and netloc in self.passwords: un, pw = self.passwords[netloc] # It is possible that the cached credentials are for a different username, # in which case the cache should be ignored. if username is None or username == un: username, password = un, pw if username is not None or password is not None: # Convert the username and password if they're None, so that # this netloc will show up as "cached" in the conditional above. # Further, HTTPBasicAuth doesn't accept None, so it makes sense to # cache the value that is going to be used. username = username or "" password = password or "" # Store any acquired credentials. self.passwords[netloc] = (username, password) assert ( # Credentials were found (username is not None and password is not None) # Credentials were not found or (username is None and password is None) ), f"Could not load credentials from url: {original_url}" return url, username, password def __call__(self, req: Request) -> Request: # Get credentials for this request url, username, password = self._get_url_and_credentials(req.url) # Set the url of the request to the url without any credentials req.url = url if username is not None and password is not None: # Send the basic auth with this request req = HTTPBasicAuth(username, password)(req) # Attach a hook to handle 401 responses req.register_hook("response", self.handle_401) return req # Factored out to allow for easy patching in tests def _prompt_for_password(self, netloc: str) -> tuple[str \| None, str \| None, bool]: username = ask_input(f"User for {netloc}: ") if self.prompting else None if not username: return None, None, False if self.use_keyring: auth = self._get_keyring_auth(netloc, username) if auth and auth[0] is not None and auth[1] is not None: return auth[0], auth[1], False password = ask_password("Password: ") return username, password, True # Factored out to allow for easy patching in tests def _should_save_password_to_keyring(self) -> bool: if ( not self.prompting or not self.use_keyring or not self.keyring_provider.has_keyring ): return False return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y" def handle_401(self, resp: Response, kwargs: Any) -> Response: # We only care about 401 responses, anything else we want to just # pass through the actual response if resp.status_code != 401: return resp username, password = None, None # Query the keyring for credentials: if self.use_keyring: username, password = self._get_new_credentials( resp.url, allow_netrc=False, allow_keyring=True, ) # We are not able to prompt the user so simply return the response if not self.prompting and not username and not password: return resp parsed = urllib.parse.urlparse(resp.url) # Prompt the user for a new username and password save = False if not username and not password: username, password, save = self._prompt_for_password(parsed.netloc) # Store the new username and password to use for future requests self._credentials_to_save = None if username is not None and password is not None: self.passwords[parsed.netloc] = (username, password) # Prompt to save the password to keyring if save and self._should_save_password_to_keyring(): self._credentials_to_save = Credentials( url=parsed.netloc, username=username, password=password, ) # Consume content and release the original connection to allow our new # request to reuse the same one. # The result of the assignment isn't used, it's just needed to consume # the content. _ = resp.content resp.raw.release_conn() # Add our new username and password to the request req = HTTPBasicAuth(username or "", password or "")(resp.request) req.register_hook("response", self.warn_on_401) # On successful request, save the credentials that were used to # keyring. (Note that if the user responded "no" above, this member # is not set and nothing will be saved.) if self._credentials_to_save: req.register_hook("response", self.save_credentials) # Send our new request new_resp = resp.connection.send(req, kwargs) new_resp.history.append(resp) return new_resp def warn_on_401(self, resp: Response, kwargs: Any) -> None: """Response callback to warn about incorrect credentials.""" if resp.status_code == 401: logger.warning( "401 Error, Credentials not correct for %s", resp.request.url, ) def save_credentials(self, resp: Response, *kwargs: Any) -> None: """Response callback to save credentials on success.""" assert ( self.keyring_provider.has_keyring ), "should never reach here without keyring" creds = self._credentials_to_save self._credentials_to_save = None if creds and resp.status_code < 400: try: logger.info("Saving credentials to keyring") self.keyring_provider.save_auth_info( creds.url, creds.username, creds.password ) except Exception: logger.exception("Failed to save credentials")	20.2 KB	2026-01-20 12:58
📄 cache.py """HTTP cache implementation.""" from __future__ import annotations import os import shutil from collections.abc import Generator from contextlib import contextmanager from datetime import datetime from typing import Any, BinaryIO, Callable from pip._vendor.cachecontrol.cache import SeparateBodyBaseCache from pip._vendor.cachecontrol.caches import SeparateBodyFileCache from pip._vendor.requests.models import Response from pip._internal.utils.filesystem import ( adjacent_tmp_file, copy_directory_permissions, replace, ) from pip._internal.utils.misc import ensure_dir def is_from_cache(response: Response) -> bool: return getattr(response, "from_cache", False) @contextmanager def suppressed_cache_errors() -> Generator[None, None, None]: """If we can't access the cache then we can just skip caching and process requests as if caching wasn't enabled. """ try: yield except OSError: pass class SafeFileCache(SeparateBodyBaseCache): """ A file based cache which is safe to use even when the target directory may not be accessible or writable. There is a race condition when two processes try to write and/or read the same entry at the same time, since each entry consists of two separate files (https://github.com/psf/cachecontrol/issues/324). We therefore have additional logic that makes sure that both files to be present before returning an entry; this fixes the read side of the race condition. For the write side, we assume that the server will only ever return the same data for the same URL, which ought to be the case for files pip is downloading. PyPI does not have a mechanism to swap out a wheel for another wheel, for example. If this assumption is not true, the CacheControl issue will need to be fixed. """ def __init__(self, directory: str) -> None: assert directory is not None, "Cache directory must not be None." super().__init__() self.directory = directory def _get_cache_path(self, name: str) -> str: # From cachecontrol.caches.file_cache.FileCache._fn, brought into our # class for backwards-compatibility and to avoid using a non-public # method. hashed = SeparateBodyFileCache.encode(name) parts = list(hashed[:5]) + [hashed] return os.path.join(self.directory, *parts) def get(self, key: str) -> bytes \| None: # The cache entry is only valid if both metadata and body exist. metadata_path = self._get_cache_path(key) body_path = metadata_path + ".body" if not (os.path.exists(metadata_path) and os.path.exists(body_path)): return None with suppressed_cache_errors(): with open(metadata_path, "rb") as f: return f.read() def _write_to_file(self, path: str, writer_func: Callable[[BinaryIO], Any]) -> None: """Common file writing logic with proper permissions and atomic replacement.""" with suppressed_cache_errors(): ensure_dir(os.path.dirname(path)) with adjacent_tmp_file(path) as f: writer_func(f) # Inherit the read/write permissions of the cache directory # to enable multi-user cache use-cases. copy_directory_permissions(self.directory, f) replace(f.name, path) def _write(self, path: str, data: bytes) -> None: self._write_to_file(path, lambda f: f.write(data)) def _write_from_io(self, path: str, source_file: BinaryIO) -> None: self._write_to_file(path, lambda f: shutil.copyfileobj(source_file, f)) def set( self, key: str, value: bytes, expires: int \| datetime \| None = None ) -> None: path = self._get_cache_path(key) self._write(path, value) def delete(self, key: str) -> None: path = self._get_cache_path(key) with suppressed_cache_errors(): os.remove(path) with suppressed_cache_errors(): os.remove(path + ".body") def get_body(self, key: str) -> BinaryIO \| None: # The cache entry is only valid if both metadata and body exist. metadata_path = self._get_cache_path(key) body_path = metadata_path + ".body" if not (os.path.exists(metadata_path) and os.path.exists(body_path)): return None with suppressed_cache_errors(): return open(body_path, "rb") def set_body(self, key: str, body: bytes) -> None: path = self._get_cache_path(key) + ".body" self._write(path, body) def set_body_from_io(self, key: str, body_file: BinaryIO) -> None: """Set the body of the cache entry from a file object.""" path = self._get_cache_path(key) + ".body" self._write_from_io(path, body_file)	4.75 KB	2026-01-20 12:58
📄 download.py """Download files with progress indicators.""" from __future__ import annotations import email.message import logging import mimetypes import os from collections.abc import Iterable, Mapping from dataclasses import dataclass from http import HTTPStatus from typing import BinaryIO from pip._vendor.requests import PreparedRequest from pip._vendor.requests.models import Response from pip._vendor.urllib3 import HTTPResponse as URLlib3Response from pip._vendor.urllib3._collections import HTTPHeaderDict from pip._vendor.urllib3.exceptions import ReadTimeoutError from pip._internal.cli.progress_bars import BarType, get_download_progress_renderer from pip._internal.exceptions import IncompleteDownloadError, NetworkConnectionError from pip._internal.models.index import PyPI from pip._internal.models.link import Link from pip._internal.network.cache import SafeFileCache, is_from_cache from pip._internal.network.session import CacheControlAdapter, PipSession from pip._internal.network.utils import HEADERS, raise_for_status, response_chunks from pip._internal.utils.misc import format_size, redact_auth_from_url, splitext logger = logging.getLogger(__name__) def _get_http_response_size(resp: Response) -> int \| None: try: return int(resp.headers["content-length"]) except (ValueError, KeyError, TypeError): return None def _get_http_response_etag_or_last_modified(resp: Response) -> str \| None: """ Return either the ETag or Last-Modified header (or None if neither exists). The return value can be used in an If-Range header. """ return resp.headers.get("etag", resp.headers.get("last-modified")) def _log_download( resp: Response, link: Link, progress_bar: BarType, total_length: int \| None, range_start: int \| None = 0, ) -> Iterable[bytes]: if link.netloc == PyPI.file_storage_domain: url = link.show_url else: url = link.url_without_fragment logged_url = redact_auth_from_url(url) if total_length: if range_start: logged_url = ( f"{logged_url} ({format_size(range_start)}/{format_size(total_length)})" ) else: logged_url = f"{logged_url} ({format_size(total_length)})" if is_from_cache(resp): logger.info("Using cached %s", logged_url) elif range_start: logger.info("Resuming download %s", logged_url) else: logger.info("Downloading %s", logged_url) if logger.getEffectiveLevel() > logging.INFO: show_progress = False elif is_from_cache(resp): show_progress = False elif not total_length: show_progress = True elif total_length > (512 * 1024): show_progress = True else: show_progress = False chunks = response_chunks(resp) if not show_progress: return chunks renderer = get_download_progress_renderer( bar_type=progress_bar, size=total_length, initial_progress=range_start ) return renderer(chunks) def sanitize_content_filename(filename: str) -> str: """ Sanitize the "filename" value from a Content-Disposition header. """ return os.path.basename(filename) def parse_content_disposition(content_disposition: str, default_filename: str) -> str: """ Parse the "filename" value from a Content-Disposition header, and return the default filename if the result is empty. """ m = email.message.Message() m["content-type"] = content_disposition filename = m.get_param("filename") if filename: # We need to sanitize the filename to prevent directory traversal # in case the filename contains ".." path parts. filename = sanitize_content_filename(str(filename)) return filename or default_filename def _get_http_response_filename(resp: Response, link: Link) -> str: """Get an ideal filename from the given HTTP response, falling back to the link filename if not provided. """ filename = link.filename # fallback # Have a look at the Content-Disposition header for a better guess content_disposition = resp.headers.get("content-disposition") if content_disposition: filename = parse_content_disposition(content_disposition, filename) ext: str \| None = splitext(filename)[1] if not ext: ext = mimetypes.guess_extension(resp.headers.get("content-type", "")) if ext: filename += ext if not ext and link.url != resp.url: ext = os.path.splitext(resp.url)[1] if ext: filename += ext return filename @dataclass class _FileDownload: """Stores the state of a single link download.""" link: Link output_file: BinaryIO size: int \| None bytes_received: int = 0 reattempts: int = 0 def is_incomplete(self) -> bool: return bool(self.size is not None and self.bytes_received < self.size) def write_chunk(self, data: bytes) -> None: self.bytes_received += len(data) self.output_file.write(data) def reset_file(self) -> None: """Delete any saved data and reset progress to zero.""" self.output_file.seek(0) self.output_file.truncate() self.bytes_received = 0 class Downloader: def __init__( self, session: PipSession, progress_bar: BarType, resume_retries: int, ) -> None: assert ( resume_retries >= 0 ), "Number of max resume retries must be bigger or equal to zero" self._session = session self._progress_bar = progress_bar self._resume_retries = resume_retries def batch( self, links: Iterable[Link], location: str ) -> Iterable[tuple[Link, tuple[str, str]]]: """Convenience method to download multiple links.""" for link in links: filepath, content_type = self(link, location) yield link, (filepath, content_type) def __call__(self, link: Link, location: str) -> tuple[str, str]: """Download a link and save it under location.""" resp = self._http_get(link) download_size = _get_http_response_size(resp) filepath = os.path.join(location, _get_http_response_filename(resp, link)) with open(filepath, "wb") as content_file: download = _FileDownload(link, content_file, download_size) self._process_response(download, resp) if download.is_incomplete(): self._attempt_resumes_or_redownloads(download, resp) content_type = resp.headers.get("Content-Type", "") return filepath, content_type def _process_response(self, download: _FileDownload, resp: Response) -> None: """Download and save chunks from a response.""" chunks = _log_download( resp, download.link, self._progress_bar, download.size, range_start=download.bytes_received, ) try: for chunk in chunks: download.write_chunk(chunk) except ReadTimeoutError as e: # If the download size is not known, then give up downloading the file. if download.size is None: raise e logger.warning("Connection timed out while downloading.") def _attempt_resumes_or_redownloads( self, download: _FileDownload, first_resp: Response ) -> None: """Attempt to resume/restart the download if connection was dropped.""" while download.reattempts < self._resume_retries and download.is_incomplete(): assert download.size is not None download.reattempts += 1 logger.warning( "Attempting to resume incomplete download (%s/%s, attempt %d)", format_size(download.bytes_received), format_size(download.size), download.reattempts, ) try: resume_resp = self._http_get_resume(download, should_match=first_resp) # Fallback: if the server responded with 200 (i.e., the file has # since been modified or range requests are unsupported) or any # other unexpected status, restart the download from the beginning. must_restart = resume_resp.status_code != HTTPStatus.PARTIAL_CONTENT if must_restart: download.reset_file() download.size = _get_http_response_size(resume_resp) first_resp = resume_resp self._process_response(download, resume_resp) except (ConnectionError, ReadTimeoutError, OSError): continue # No more resume attempts. Raise an error if the download is still incomplete. if download.is_incomplete(): os.remove(download.output_file.name) raise IncompleteDownloadError(download) # If we successfully completed the download via resume, manually cache it # as a complete response to enable future caching if download.reattempts > 0: self._cache_resumed_download(download, first_resp) def _cache_resumed_download( self, download: _FileDownload, original_response: Response ) -> None: """ Manually cache a file that was successfully downloaded via resume retries. cachecontrol doesn't cache 206 (Partial Content) responses, since they are not complete files. This method manually adds the final file to the cache as though it was downloaded in a single request, so that future requests can use the cache. """ url = download.link.url_without_fragment adapter = self._session.get_adapter(url) # Check if the adapter is the CacheControlAdapter (i.e. caching is enabled) if not isinstance(adapter, CacheControlAdapter): logger.debug( "Skipping resume download caching: no cache controller for %s", url ) return # Check SafeFileCache is being used assert isinstance( adapter.cache, SafeFileCache ), "separate body cache not in use!" synthetic_request = PreparedRequest() synthetic_request.prepare(method="GET", url=url, headers={}) synthetic_response_headers = HTTPHeaderDict() for key, value in original_response.headers.items(): if key.lower() not in ["content-range", "content-length"]: synthetic_response_headers[key] = value synthetic_response_headers["content-length"] = str(download.size) synthetic_response = URLlib3Response( body="", headers=synthetic_response_headers, status=200, preload_content=False, ) # Save metadata and then stream the file contents to cache. cache_url = adapter.controller.cache_url(url) metadata_blob = adapter.controller.serializer.dumps( synthetic_request, synthetic_response, b"" ) adapter.cache.set(cache_url, metadata_blob) download.output_file.flush() with open(download.output_file.name, "rb") as f: adapter.cache.set_body_from_io(cache_url, f) logger.debug( "Cached resumed download as complete response for future use: %s", url ) def _http_get_resume( self, download: _FileDownload, should_match: Response ) -> Response: """Issue a HTTP range request to resume the download.""" # To better understand the download resumption logic, see the mdn web docs: # https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests headers = HEADERS.copy() headers["Range"] = f"bytes={download.bytes_received}-" # If possible, use a conditional range request to avoid corrupted # downloads caused by the remote file changing in-between. if identifier := _get_http_response_etag_or_last_modified(should_match): headers["If-Range"] = identifier return self._http_get(download.link, headers) def _http_get(self, link: Link, headers: Mapping[str, str] = HEADERS) -> Response: target_url = link.url_without_fragment try: resp = self._session.get(target_url, headers=headers, stream=True) raise_for_status(resp) except NetworkConnectionError as e: assert e.response is not None logger.critical( "HTTP error %s while getting %s", e.response.status_code, link ) raise return resp	12.38 KB	2026-01-20 12:58
📄 lazy_wheel.py """Lazy ZIP over HTTP""" from __future__ import annotations __all__ = ["HTTPRangeRequestUnsupported", "dist_from_wheel_url"] from bisect import bisect_left, bisect_right from collections.abc import Generator from contextlib import contextmanager from tempfile import NamedTemporaryFile from typing import Any from zipfile import BadZipFile, ZipFile from pip._vendor.packaging.utils import NormalizedName from pip._vendor.requests.models import CONTENT_CHUNK_SIZE, Response from pip._internal.metadata import BaseDistribution, MemoryWheel, get_wheel_distribution from pip._internal.network.session import PipSession from pip._internal.network.utils import HEADERS, raise_for_status, response_chunks class HTTPRangeRequestUnsupported(Exception): pass def dist_from_wheel_url( name: NormalizedName, url: str, session: PipSession ) -> BaseDistribution: """Return a distribution object from the given wheel URL. This uses HTTP range requests to only fetch the portion of the wheel containing metadata, just enough for the object to be constructed. If such requests are not supported, HTTPRangeRequestUnsupported is raised. """ with LazyZipOverHTTP(url, session) as zf: # For read-only ZIP files, ZipFile only needs methods read, # seek, seekable and tell, not the whole IO protocol. wheel = MemoryWheel(zf.name, zf) # type: ignore # After context manager exit, wheel.name # is an invalid file by intention. return get_wheel_distribution(wheel, name) class LazyZipOverHTTP: """File-like object mapped to a ZIP file over HTTP. This uses HTTP range requests to lazily fetch the file's content, which is supposed to be fed to ZipFile. If such requests are not supported by the server, raise HTTPRangeRequestUnsupported during initialization. """ def __init__( self, url: str, session: PipSession, chunk_size: int = CONTENT_CHUNK_SIZE ) -> None: head = session.head(url, headers=HEADERS) raise_for_status(head) assert head.status_code == 200 self._session, self._url, self._chunk_size = session, url, chunk_size self._length = int(head.headers["Content-Length"]) self._file = NamedTemporaryFile() self.truncate(self._length) self._left: list[int] = [] self._right: list[int] = [] if "bytes" not in head.headers.get("Accept-Ranges", "none"): raise HTTPRangeRequestUnsupported("range request is not supported") self._check_zip() @property def mode(self) -> str: """Opening mode, which is always rb.""" return "rb" @property def name(self) -> str: """Path to the underlying file.""" return self._file.name def seekable(self) -> bool: """Return whether random access is supported, which is True.""" return True def close(self) -> None: """Close the file.""" self._file.close() @property def closed(self) -> bool: """Whether the file is closed.""" return self._file.closed def read(self, size: int = -1) -> bytes: """Read up to size bytes from the object and return them. As a convenience, if size is unspecified or -1, all bytes until EOF are returned. Fewer than size bytes may be returned if EOF is reached. """ download_size = max(size, self._chunk_size) start, length = self.tell(), self._length stop = length if size < 0 else min(start + download_size, length) start = max(0, stop - download_size) self._download(start, stop - 1) return self._file.read(size) def readable(self) -> bool: """Return whether the file is readable, which is True.""" return True def seek(self, offset: int, whence: int = 0) -> int: """Change stream position and return the new absolute position. Seek to offset relative position indicated by whence: * 0: Start of stream (the default). pos should be >= 0; * 1: Current position - pos may be negative; * 2: End of stream - pos usually negative. """ return self._file.seek(offset, whence) def tell(self) -> int: """Return the current position.""" return self._file.tell() def truncate(self, size: int \| None = None) -> int: """Resize the stream to the given size in bytes. If size is unspecified resize to the current position. The current stream position isn't changed. Return the new file size. """ return self._file.truncate(size) def writable(self) -> bool: """Return False.""" return False def __enter__(self) -> LazyZipOverHTTP: self._file.__enter__() return self def __exit__(self, exc: Any) -> None: self._file.__exit__(exc) @contextmanager def _stay(self) -> Generator[None, None, None]: """Return a context manager keeping the position. At the end of the block, seek back to original position. """ pos = self.tell() try: yield finally: self.seek(pos) def _check_zip(self) -> None: """Check and download until the file is a valid ZIP.""" end = self._length - 1 for start in reversed(range(0, end, self._chunk_size)): self._download(start, end) with self._stay(): try: # For read-only ZIP files, ZipFile only needs # methods read, seek, seekable and tell. ZipFile(self) except BadZipFile: pass else: break def _stream_response( self, start: int, end: int, base_headers: dict[str, str] = HEADERS ) -> Response: """Return HTTP response to a range request from start to end.""" headers = base_headers.copy() headers["Range"] = f"bytes={start}-{end}" # TODO: Get range requests to be correctly cached headers["Cache-Control"] = "no-cache" return self._session.get(self._url, headers=headers, stream=True) def _merge( self, start: int, end: int, left: int, right: int ) -> Generator[tuple[int, int], None, None]: """Return a generator of intervals to be fetched. Args: start (int): Start of needed interval end (int): End of needed interval left (int): Index of first overlapping downloaded data right (int): Index after last overlapping downloaded data """ lslice, rslice = self._left[left:right], self._right[left:right] i = start = min([start] + lslice[:1]) end = max([end] + rslice[-1:]) for j, k in zip(lslice, rslice): if j > i: yield i, j - 1 i = k + 1 if i <= end: yield i, end self._left[left:right], self._right[left:right] = [start], [end] def _download(self, start: int, end: int) -> None: """Download bytes from start to end inclusively.""" with self._stay(): left = bisect_left(self._right, start) right = bisect_right(self._left, end) for start, end in self._merge(start, end, left, right): response = self._stream_response(start, end) response.raise_for_status() self.seek(start) for chunk in response_chunks(response, self._chunk_size): self._file.write(chunk)	7.47 KB	2026-01-20 12:58
📄 session.py """PipSession and supporting code, containing all pip-specific network request configuration and behavior. """ from __future__ import annotations import email.utils import functools import io import ipaddress import json import logging import mimetypes import os import platform import shutil import subprocess import sys import urllib.parse import warnings from collections.abc import Generator, Mapping, Sequence from typing import ( TYPE_CHECKING, Any, Optional, Union, ) from pip._vendor import requests, urllib3 from pip._vendor.cachecontrol import CacheControlAdapter as _BaseCacheControlAdapter from pip._vendor.requests.adapters import DEFAULT_POOLBLOCK, BaseAdapter from pip._vendor.requests.adapters import HTTPAdapter as _BaseHTTPAdapter from pip._vendor.requests.models import PreparedRequest, Response from pip._vendor.requests.structures import CaseInsensitiveDict from pip._vendor.urllib3.connectionpool import ConnectionPool from pip._vendor.urllib3.exceptions import InsecureRequestWarning from pip import __version__ from pip._internal.metadata import get_default_environment from pip._internal.models.link import Link from pip._internal.network.auth import MultiDomainBasicAuth from pip._internal.network.cache import SafeFileCache # Import ssl from compat so the initial import occurs in only one place. from pip._internal.utils.compat import has_tls from pip._internal.utils.glibc import libc_ver from pip._internal.utils.misc import build_url_from_netloc, parse_netloc from pip._internal.utils.urls import url_to_path if TYPE_CHECKING: from ssl import SSLContext from pip._vendor.urllib3.poolmanager import PoolManager from pip._vendor.urllib3.proxymanager import ProxyManager logger = logging.getLogger(__name__) SecureOrigin = tuple[str, str, Optional[Union[int, str]]] # Ignore warning raised when using --trusted-host. warnings.filterwarnings("ignore", category=InsecureRequestWarning) SECURE_ORIGINS: list[SecureOrigin] = [ # protocol, hostname, port # Taken from Chrome's list of secure origins (See: http://bit.ly/1qrySKC) ("https", "", ""), ("", "localhost", ""), ("", "127.0.0.0/8", ""), ("", "::1/128", ""), ("file", "", None), # ssh is always secure. ("ssh", "", ""), ] # These are environment variables present when running under various # CI systems. For each variable, some CI systems that use the variable # are indicated. The collection was chosen so that for each of a number # of popular systems, at least one of the environment variables is used. # This list is used to provide some indication of and lower bound for # CI traffic to PyPI. Thus, it is okay if the list is not comprehensive. # For more background, see: https://github.com/pypa/pip/issues/5499 CI_ENVIRONMENT_VARIABLES = ( # Azure Pipelines "BUILD_BUILDID", # Jenkins "BUILD_ID", # AppVeyor, CircleCI, Codeship, Gitlab CI, Shippable, Travis CI "CI", # Explicit environment variable. "PIP_IS_CI", ) def looks_like_ci() -> bool: """ Return whether it looks like pip is running under CI. """ # We don't use the method of checking for a tty (e.g. using isatty()) # because some CI systems mimic a tty (e.g. Travis CI). Thus that # method doesn't provide definitive information in either direction. return any(name in os.environ for name in CI_ENVIRONMENT_VARIABLES) @functools.lru_cache(maxsize=1) def user_agent() -> str: """ Return a string representing the user agent. """ data: dict[str, Any] = { "installer": {"name": "pip", "version": __version__}, "python": platform.python_version(), "implementation": { "name": platform.python_implementation(), }, } if data["implementation"]["name"] == "CPython": data["implementation"]["version"] = platform.python_version() elif data["implementation"]["name"] == "PyPy": pypy_version_info = sys.pypy_version_info # type: ignore if pypy_version_info.releaselevel == "final": pypy_version_info = pypy_version_info[:3] data["implementation"]["version"] = ".".join( [str(x) for x in pypy_version_info] ) elif data["implementation"]["name"] == "Jython": # Complete Guess data["implementation"]["version"] = platform.python_version() elif data["implementation"]["name"] == "IronPython": # Complete Guess data["implementation"]["version"] = platform.python_version() if sys.platform.startswith("linux"): from pip._vendor import distro linux_distribution = distro.name(), distro.version(), distro.codename() distro_infos: dict[str, Any] = dict( filter( lambda x: x[1], zip(["name", "version", "id"], linux_distribution), ) ) libc = dict( filter( lambda x: x[1], zip(["lib", "version"], libc_ver()), ) ) if libc: distro_infos["libc"] = libc if distro_infos: data["distro"] = distro_infos if sys.platform.startswith("darwin") and platform.mac_ver()[0]: data["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]} if platform.system(): data.setdefault("system", {})["name"] = platform.system() if platform.release(): data.setdefault("system", {})["release"] = platform.release() if platform.machine(): data["cpu"] = platform.machine() if has_tls(): import _ssl as ssl data["openssl_version"] = ssl.OPENSSL_VERSION setuptools_dist = get_default_environment().get_distribution("setuptools") if setuptools_dist is not None: data["setuptools_version"] = str(setuptools_dist.version) if shutil.which("rustc") is not None: # If for any reason `rustc --version` fails, silently ignore it try: rustc_output = subprocess.check_output( ["rustc", "--version"], stderr=subprocess.STDOUT, timeout=0.5 ) except Exception: pass else: if rustc_output.startswith(b"rustc "): # The format of `rustc --version` is: # `b'rustc 1.52.1 (9bc8c42bb 2021-05-09)\n'` # We extract just the middle (1.52.1) part data["rustc_version"] = rustc_output.split(b" ")[1].decode() # Use None rather than False so as not to give the impression that # pip knows it is not being run under CI. Rather, it is a null or # inconclusive result. Also, we include some value rather than no # value to make it easier to know that the check has been run. data["ci"] = True if looks_like_ci() else None user_data = os.environ.get("PIP_USER_AGENT_USER_DATA") if user_data is not None: data["user_data"] = user_data return "{data[installer][name]}/{data[installer][version]} {json}".format( data=data, json=json.dumps(data, separators=(",", ":"), sort_keys=True), ) class LocalFSAdapter(BaseAdapter): def send( self, request: PreparedRequest, stream: bool = False, timeout: float \| tuple[float, float] \| None = None, verify: bool \| str = True, cert: str \| tuple[str, str] \| None = None, proxies: Mapping[str, str] \| None = None, ) -> Response: pathname = url_to_path(request.url) resp = Response() resp.status_code = 200 resp.url = request.url try: stats = os.stat(pathname) except OSError as exc: # format the exception raised as a io.BytesIO object, # to return a better error message: resp.status_code = 404 resp.reason = type(exc).__name__ resp.raw = io.BytesIO(f"{resp.reason}: {exc}".encode()) else: modified = email.utils.formatdate(stats.st_mtime, usegmt=True) content_type = mimetypes.guess_type(pathname)[0] or "text/plain" resp.headers = CaseInsensitiveDict( { "Content-Type": content_type, "Content-Length": stats.st_size, "Last-Modified": modified, } ) resp.raw = open(pathname, "rb") resp.close = resp.raw.close return resp def close(self) -> None: pass class _SSLContextAdapterMixin: """Mixin to add the ``ssl_context`` constructor argument to HTTP adapters. The additional argument is forwarded directly to the pool manager. This allows us to dynamically decide what SSL store to use at runtime, which is used to implement the optional ``truststore`` backend. """ def __init__( self, , ssl_context: SSLContext \| None = None, kwargs: Any, ) -> None: self._ssl_context = ssl_context super().__init__(kwargs) def init_poolmanager( self, connections: int, maxsize: int, block: bool = DEFAULT_POOLBLOCK, pool_kwargs: Any, ) -> PoolManager: if self._ssl_context is not None: pool_kwargs.setdefault("ssl_context", self._ssl_context) return super().init_poolmanager( # type: ignore[misc] connections=connections, maxsize=maxsize, block=block, pool_kwargs, ) def proxy_manager_for(self, proxy: str, proxy_kwargs: Any) -> ProxyManager: # Proxy manager replaces the pool manager, so inject our SSL # context here too. https://github.com/pypa/pip/issues/13288 if self._ssl_context is not None: proxy_kwargs.setdefault("ssl_context", self._ssl_context) return super().proxy_manager_for(proxy, proxy_kwargs) # type: ignore[misc] class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter): pass class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter): pass class InsecureHTTPAdapter(HTTPAdapter): def cert_verify( self, conn: ConnectionPool, url: str, verify: bool \| str, cert: str \| tuple[str, str] \| None, ) -> None: super().cert_verify(conn=conn, url=url, verify=False, cert=cert) class InsecureCacheControlAdapter(CacheControlAdapter): def cert_verify( self, conn: ConnectionPool, url: str, verify: bool \| str, cert: str \| tuple[str, str] \| None, ) -> None: super().cert_verify(conn=conn, url=url, verify=False, cert=cert) class PipSession(requests.Session): timeout: int \| None = None def __init__( self, args: Any, retries: int = 0, cache: str \| None = None, trusted_hosts: Sequence[str] = (), index_urls: list[str] \| None = None, ssl_context: SSLContext \| None = None, kwargs: Any, ) -> None: """ :param trusted_hosts: Domains not to emit warnings for when not using HTTPS. """ super().__init__(args, *kwargs) # Namespace the attribute with "pip_" just in case to prevent # possible conflicts with the base class. self.pip_trusted_origins: list[tuple[str, int \| None]] = [] self.pip_proxy = None # Attach our User Agent to the request self.headers["User-Agent"] = user_agent() # Attach our Authentication handler to the session self.auth = MultiDomainBasicAuth(index_urls=index_urls) # Create our urllib3.Retry instance which will allow us to customize # how we handle retries. retries = urllib3.Retry( # Set the total number of retries that a particular request can # have. total=retries, # A 503 error from PyPI typically means that the Fastly -> Origin # connection got interrupted in some way. A 503 error in general # is typically considered a transient error so we'll go ahead and # retry it. # A 500 may indicate transient error in Amazon S3 # A 502 may be a transient error from a CDN like CloudFlare or CloudFront # A 520 or 527 - may indicate transient error in CloudFlare status_forcelist=[500, 502, 503, 520, 527], # Add a small amount of back off between failed requests in # order to prevent hammering the service. backoff_factor=0.25, ) # type: ignore # Our Insecure HTTPAdapter disables HTTPS validation. It does not # support caching so we'll use it for all http:// URLs. # If caching is disabled, we will also use it for # https:// hosts that we've marked as ignoring # TLS errors for (trusted-hosts). insecure_adapter = InsecureHTTPAdapter(max_retries=retries) # We want to _only_ cache responses on securely fetched origins or when # the host is specified as trusted. We do this because # we can't validate the response of an insecurely/untrusted fetched # origin, and we don't want someone to be able to poison the cache and # require manual eviction from the cache to fix it. if cache: secure_adapter = CacheControlAdapter( cache=SafeFileCache(cache), max_retries=retries, ssl_context=ssl_context, ) self._trusted_host_adapter = InsecureCacheControlAdapter( cache=SafeFileCache(cache), max_retries=retries, ) else: secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context) self._trusted_host_adapter = insecure_adapter self.mount("https://", secure_adapter) self.mount("http://", insecure_adapter) # Enable file:// urls self.mount("file://", LocalFSAdapter()) for host in trusted_hosts: self.add_trusted_host(host, suppress_logging=True) def update_index_urls(self, new_index_urls: list[str]) -> None: """ :param new_index_urls: New index urls to update the authentication handler with. """ self.auth.index_urls = new_index_urls def add_trusted_host( self, host: str, source: str \| None = None, suppress_logging: bool = False ) -> None: """ :param host: It is okay to provide a host that has previously been added. :param source: An optional source string, for logging where the host string came from. """ if not suppress_logging: msg = f"adding trusted host: {host!r}" if source is not None: msg += f" (from {source})" logger.info(msg) parsed_host, parsed_port = parse_netloc(host) if parsed_host is None: raise ValueError(f"Trusted host URL must include a host part: {host!r}") if (parsed_host, parsed_port) not in self.pip_trusted_origins: self.pip_trusted_origins.append((parsed_host, parsed_port)) self.mount( build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter ) self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter) if not parsed_port: self.mount( build_url_from_netloc(host, scheme="http") + ":", self._trusted_host_adapter, ) # Mount wildcard ports for the same host. self.mount(build_url_from_netloc(host) + ":", self._trusted_host_adapter) def iter_secure_origins(self) -> Generator[SecureOrigin, None, None]: yield from SECURE_ORIGINS for host, port in self.pip_trusted_origins: yield ("", host, "" if port is None else port) def is_secure_origin(self, location: Link) -> bool: # Determine if this url used a secure transport mechanism parsed = urllib.parse.urlparse(str(location)) origin_protocol, origin_host, origin_port = ( parsed.scheme, parsed.hostname, parsed.port, ) # The protocol to use to see if the protocol matches. # Don't count the repository type as part of the protocol: in # cases such as "git+ssh", only use "ssh". (I.e., Only verify against # the last scheme.) origin_protocol = origin_protocol.rsplit("+", 1)[-1] # Determine if our origin is a secure origin by looking through our # hardcoded list of secure origins, as well as any additional ones # configured on this PackageFinder instance. for secure_origin in self.iter_secure_origins(): secure_protocol, secure_host, secure_port = secure_origin if origin_protocol != secure_protocol and secure_protocol != "": continue try: addr = ipaddress.ip_address(origin_host or "") network = ipaddress.ip_network(secure_host) except ValueError: # We don't have both a valid address or a valid network, so # we'll check this origin against hostnames. if ( origin_host and origin_host.lower() != secure_host.lower() and secure_host != "" ): continue else: # We have a valid address and network, so see if the address # is contained within the network. if addr not in network: continue # Check to see if the port matches. if ( origin_port != secure_port and secure_port != "" and secure_port is not None ): continue # If we've gotten here, then this origin matches the current # secure origin and we should return True return True # If we've gotten to this point, then the origin isn't secure and we # will not accept it as a valid location to search. We will however # log a warning that we are ignoring it. logger.warning( "The repository located at %s is not a trusted or secure host and " "is being ignored. If this repository is available via HTTPS we " "recommend you use HTTPS instead, otherwise you may silence " "this warning and allow it anyway with '--trusted-host %s'.", origin_host, origin_host, ) return False def request(self, method: str, url: str, args: Any, kwargs: Any) -> Response: # Allow setting a default timeout on a session kwargs.setdefault("timeout", self.timeout) # Allow setting a default proxies on a session kwargs.setdefault("proxies", self.proxies) # Dispatch the actual request return super().request(method, url, args, **kwargs)	18.74 KB	2026-01-20 12:58
📄 utils.py from collections.abc import Generator from pip._vendor.requests.models import Response from pip._internal.exceptions import NetworkConnectionError # The following comments and HTTP headers were originally added by # Donald Stufft in git commit 22c562429a61bb77172039e480873fb239dd8c03. # # We use Accept-Encoding: identity here because requests defaults to # accepting compressed responses. This breaks in a variety of ways # depending on how the server is configured. # - Some servers will notice that the file isn't a compressible file # and will leave the file alone and with an empty Content-Encoding # - Some servers will notice that the file is already compressed and # will leave the file alone, adding a Content-Encoding: gzip header # - Some servers won't notice anything at all and will take a file # that's already been compressed and compress it again, and set # the Content-Encoding: gzip header # By setting this to request only the identity encoding we're hoping # to eliminate the third case. Hopefully there does not exist a server # which when given a file will notice it is already compressed and that # you're not asking for a compressed file and will then decompress it # before sending because if that's the case I don't think it'll ever be # possible to make this work. HEADERS: dict[str, str] = {"Accept-Encoding": "identity"} DOWNLOAD_CHUNK_SIZE = 256 * 1024 def raise_for_status(resp: Response) -> None: http_error_msg = "" if isinstance(resp.reason, bytes): # We attempt to decode utf-8 first because some servers # choose to localize their reason strings. If the string # isn't utf-8, we fall back to iso-8859-1 for all other # encodings. try: reason = resp.reason.decode("utf-8") except UnicodeDecodeError: reason = resp.reason.decode("iso-8859-1") else: reason = resp.reason if 400 <= resp.status_code < 500: http_error_msg = ( f"{resp.status_code} Client Error: {reason} for url: {resp.url}" ) elif 500 <= resp.status_code < 600: http_error_msg = ( f"{resp.status_code} Server Error: {reason} for url: {resp.url}" ) if http_error_msg: raise NetworkConnectionError(http_error_msg, response=resp) def response_chunks( response: Response, chunk_size: int = DOWNLOAD_CHUNK_SIZE ) -> Generator[bytes, None, None]: """Given a requests Response, provide the data chunks.""" try: # Special case for urllib3. for chunk in response.raw.stream( chunk_size, # We use decode_content=False here because we don't # want urllib3 to mess with the raw bytes we get # from the server. If we decompress inside of # urllib3 then we cannot verify the checksum # because the checksum will be of the compressed # file. This breakage will only occur if the # server adds a Content-Encoding header, which # depends on how the server was configured: # - Some servers will notice that the file isn't a # compressible file and will leave the file alone # and with an empty Content-Encoding # - Some servers will notice that the file is # already compressed and will leave the file # alone and will add a Content-Encoding: gzip # header # - Some servers won't notice anything at all and # will take a file that's already been compressed # and compress it again and set the # Content-Encoding: gzip header # # By setting this not to decode automatically we # hope to eliminate problems with the second case. decode_content=False, ): yield chunk except AttributeError: # Standard file-like object. while True: chunk = response.raw.read(chunk_size) if not chunk: break yield chunk	4 KB	2026-01-20 12:58
📄 xmlrpc.py """xmlrpclib.Transport implementation""" import logging import urllib.parse import xmlrpc.client from typing import TYPE_CHECKING from pip._internal.exceptions import NetworkConnectionError from pip._internal.network.session import PipSession from pip._internal.network.utils import raise_for_status if TYPE_CHECKING: from xmlrpc.client import _HostType, _Marshallable from _typeshed import SizedBuffer logger = logging.getLogger(__name__) class PipXmlrpcTransport(xmlrpc.client.Transport): """Provide a `xmlrpclib.Transport` implementation via a `PipSession` object. """ def __init__( self, index_url: str, session: PipSession, use_datetime: bool = False ) -> None: super().__init__(use_datetime) index_parts = urllib.parse.urlparse(index_url) self._scheme = index_parts.scheme self._session = session def request( self, host: "_HostType", handler: str, request_body: "SizedBuffer", verbose: bool = False, ) -> tuple["_Marshallable", ...]: assert isinstance(host, str) parts = (self._scheme, host, handler, None, None, None) url = urllib.parse.urlunparse(parts) try: headers = {"Content-Type": "text/xml"} response = self._session.post( url, data=request_body, headers=headers, stream=True, ) raise_for_status(response) self.verbose = verbose return self.parse_response(response.raw) except NetworkConnectionError as exc: assert exc.response logger.critical( "HTTP error %s while getting %s", exc.response.status_code, url, ) raise	1.79 KB	2026-01-20 12:58

Terminal