Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import datetime
4import functools
5import itertools
6import logging
7import os
8import posixpath
9import re
10import urllib.parse
11from collections.abc import Mapping
12from dataclasses import dataclass
13from typing import (
14 Any,
15 NamedTuple,
16 NewType,
17)
19from pip._internal.exceptions import InvalidEggFragment
20from pip._internal.utils.datetime import parse_iso_datetime
21from pip._internal.utils.filetypes import WHEEL_EXTENSION
22from pip._internal.utils.hashes import Hashes
23from pip._internal.utils.misc import (
24 pairwise,
25 redact_auth_from_url,
26 split_auth_from_netloc,
27 splitext,
28)
29from pip._internal.utils.urls import path_to_url, url_to_path
31logger = logging.getLogger(__name__)
34# A single path component: percent-decoded once and reduced to a basename, so it
35# contains no path separator and is not a ``.`` or ``..`` reference. The empty
36# string means "no component".
37PathComponent = NewType("PathComponent", str)
40def _to_path_component(name: str) -> PathComponent:
41 """Reduce ``name`` to a single path component, or ``""`` if it has none.
43 ``os.path.basename`` drops any directory part, drive letter, or separator;
44 a ``.``, ``..``, or empty result is not a component and becomes ``""``.
45 """
46 name = os.path.basename(name)
47 if name in ("", os.curdir, os.pardir):
48 return PathComponent("")
50 return PathComponent(name)
53def as_path_component(name: str) -> PathComponent:
54 """Like ``_to_path_component`` but reject the empty result.
56 Use where a file is about to be written, so a missing name is an error
57 rather than a silent fallback to the directory itself.
58 """
59 component = _to_path_component(name)
60 if not component:
61 raise ValueError(f"Unexpected file name derived from URL: {name!r}")
63 return component
66def join_within_directory(directory: str, component: PathComponent) -> str:
67 """Join a single path ``component`` onto ``directory``.
69 ``component`` is a :data:`PathComponent`, so by type it has no separator and
70 is not a ``.`` or ``..`` reference; the result can never escape ``directory``.
71 Requiring ``PathComponent`` rather than ``str`` lets the type checker enforce
72 at the call site that the name was reduced to a safe component beforehand.
73 """
74 return os.path.join(directory, component)
77# Order matters, earlier hashes have a precedence over later hashes for what
78# we will pick to use.
79_SUPPORTED_HASHES = ("sha512", "sha384", "sha256", "sha224", "sha1", "md5")
82@dataclass(frozen=True)
83class LinkHash:
84 """Links to content may have embedded hash values. This class parses those.
86 `name` must be any member of `_SUPPORTED_HASHES`.
88 This class can be converted to and from `ArchiveInfo`. While ArchiveInfo intends to
89 be JSON-serializable to conform to PEP 610, this class contains the logic for
90 parsing a hash name and value for correctness, and then checking whether that hash
91 conforms to a schema with `.is_hash_allowed()`."""
93 name: str
94 value: str
96 _hash_url_fragment_re = re.compile(
97 # NB: we do not validate that the second group (.*) is a valid hex
98 # digest. Instead, we simply keep that string in this class, and then check it
99 # against Hashes when hash-checking is needed. This is easier to debug than
100 # proactively discarding an invalid hex digest, as we handle incorrect hashes
101 # and malformed hashes in the same place.
102 r"[#&]({choices})=([^&]*)".format(
103 choices="|".join(re.escape(hash_name) for hash_name in _SUPPORTED_HASHES)
104 ),
105 )
107 def __post_init__(self) -> None:
108 assert self.name in _SUPPORTED_HASHES
110 @classmethod
111 @functools.cache
112 def find_hash_url_fragment(cls, url: str) -> LinkHash | None:
113 """Search a string for a checksum algorithm name and encoded output value."""
114 match = cls._hash_url_fragment_re.search(url)
115 if match is None:
116 return None
117 name, value = match.groups()
118 return cls(name=name, value=value)
120 def as_dict(self) -> dict[str, str]:
121 return {self.name: self.value}
123 def as_hashes(self) -> Hashes:
124 """Return a Hashes instance which checks only for the current hash."""
125 return Hashes({self.name: [self.value]})
127 def is_hash_allowed(self, hashes: Hashes | None) -> bool:
128 """
129 Return True if the current hash is allowed by `hashes`.
130 """
131 if hashes is None:
132 return False
133 return hashes.is_hash_allowed(self.name, hex_digest=self.value)
136@dataclass(frozen=True)
137class MetadataFile:
138 """Information about a core metadata file associated with a distribution."""
140 hashes: dict[str, str] | None
142 def __post_init__(self) -> None:
143 if self.hashes is not None:
144 assert all(name in _SUPPORTED_HASHES for name in self.hashes)
147def supported_hashes(hashes: dict[str, str] | None) -> dict[str, str] | None:
148 # Remove any unsupported hash types from the mapping. If this leaves no
149 # supported hashes, return None
150 if hashes is None:
151 return None
152 hashes = {n: v for n, v in hashes.items() if n in _SUPPORTED_HASHES}
153 if not hashes:
154 return None
155 return hashes
158def _clean_url_path_part(part: str) -> str:
159 """
160 Clean a "part" of a URL path (i.e. after splitting on "@" characters).
161 """
162 # We unquote prior to quoting to make sure nothing is double quoted.
163 return urllib.parse.quote(urllib.parse.unquote(part))
166def _clean_file_url_path(part: str) -> str:
167 """
168 Clean the first part of a URL path that corresponds to a local
169 filesystem path (i.e. the first part after splitting on "@" characters).
170 """
171 import urllib.request
173 # We unquote prior to quoting to make sure nothing is double quoted.
174 # Also, on Windows the path part might contain a drive letter which
175 # should not be quoted. On Linux where drive letters do not
176 # exist, the colon should be quoted. We rely on urllib.request
177 # to do the right thing here.
178 ret = urllib.request.pathname2url(urllib.request.url2pathname(part))
179 if ret.startswith("///"):
180 # Remove any URL authority section, leaving only the URL path.
181 ret = ret.removeprefix("//")
182 return ret
185# percent-encoded: /
186_reserved_chars_re = re.compile("(@|%2F)", re.IGNORECASE)
189def _clean_url_path(path: str, is_local_path: bool) -> str:
190 """
191 Clean the path portion of a URL.
192 """
193 if is_local_path:
194 clean_func = _clean_file_url_path
195 else:
196 clean_func = _clean_url_path_part
198 # Split on the reserved characters prior to cleaning so that
199 # revision strings in VCS URLs are properly preserved.
200 parts = _reserved_chars_re.split(path)
202 cleaned_parts = []
203 for to_clean, reserved in pairwise(itertools.chain(parts, [""])):
204 cleaned_parts.append(clean_func(to_clean))
205 # Normalize %xx escapes (e.g. %2f -> %2F)
206 cleaned_parts.append(reserved.upper())
208 return "".join(cleaned_parts)
211def _ensure_quoted_url(url: str) -> str:
212 """
213 Make sure a link is fully quoted.
214 For example, if ' ' occurs in the URL, it will be replaced with "%20",
215 and without double-quoting other characters.
216 """
217 # Split the URL into parts according to the general structure
218 # `scheme://netloc/path?query#fragment`.
219 result = urllib.parse.urlsplit(url)
220 # If the netloc is empty, then the URL refers to a local filesystem path.
221 is_local_path = not result.netloc
222 path = _clean_url_path(result.path, is_local_path=is_local_path)
223 # Temporarily replace scheme with file to ensure the URL generated by
224 # urlunsplit() contains an empty netloc (file://) as per RFC 1738.
225 ret = urllib.parse.urlunsplit(result._replace(scheme="file", path=path))
226 ret = result.scheme + ret[4:] # Restore original scheme.
227 return ret
230def _absolute_link_url(base_url: str, url: str) -> str:
231 """
232 A faster implementation of urllib.parse.urljoin with a shortcut
233 for absolute http/https URLs.
234 """
235 if url.startswith(("https://", "http://")):
236 return url
237 else:
238 return urllib.parse.urljoin(base_url, url)
241@functools.total_ordering
242class Link:
243 """Represents a parsed link from a Package Index's simple URL"""
245 __slots__ = [
246 "_parsed_url",
247 "_url",
248 "_path",
249 "_hashes",
250 "comes_from",
251 "requires_python",
252 "yanked_reason",
253 "metadata_file_data",
254 "upload_time",
255 "cache_link_parsing",
256 "egg_fragment",
257 ]
259 def __init__(
260 self,
261 url: str,
262 comes_from: str | None = None,
263 requires_python: str | None = None,
264 yanked_reason: str | None = None,
265 metadata_file_data: MetadataFile | None = None,
266 upload_time: datetime.datetime | None = None,
267 cache_link_parsing: bool = True,
268 hashes: Mapping[str, str] | None = None,
269 ) -> None:
270 """
271 :param url: url of the resource pointed to (href of the link)
272 :param comes_from: URL or string indicating where the link was found.
273 :param requires_python: String containing the `Requires-Python`
274 metadata field, specified in PEP 345. This may be specified by
275 a data-requires-python attribute in the HTML link tag, as
276 described in PEP 503.
277 :param yanked_reason: the reason the file has been yanked, if the
278 file has been yanked, or None if the file hasn't been yanked.
279 This is the value of the "data-yanked" attribute, if present, in
280 a simple repository HTML link. If the file has been yanked but
281 no reason was provided, this should be the empty string. See
282 PEP 592 for more information and the specification.
283 :param metadata_file_data: the metadata attached to the file, or None if
284 no such metadata is provided. This argument, if not None, indicates
285 that a separate metadata file exists, and also optionally supplies
286 hashes for that file.
287 :param upload_time: upload time of the file, or None if the information
288 is not available from the server.
289 :param cache_link_parsing: A flag that is used elsewhere to determine
290 whether resources retrieved from this link should be cached. PyPI
291 URLs should generally have this set to False, for example.
292 :param hashes: A mapping of hash names to digests to allow us to
293 determine the validity of a download.
294 """
296 # The comes_from, requires_python, and metadata_file_data arguments are
297 # only used by classmethods of this class, and are not used in client
298 # code directly.
300 # url can be a UNC windows share
301 if url.startswith("\\\\"):
302 url = path_to_url(url)
304 self._parsed_url = urllib.parse.urlsplit(url)
305 # Store the url as a private attribute to prevent accidentally
306 # trying to set a new value.
307 self._url = url
308 # The .path property is hot, so calculate its value ahead of time.
309 self._path = urllib.parse.unquote(self._parsed_url.path)
311 link_hash = LinkHash.find_hash_url_fragment(url)
312 hashes_from_link = {} if link_hash is None else link_hash.as_dict()
313 if hashes is None:
314 self._hashes = hashes_from_link
315 else:
316 self._hashes = {**hashes, **hashes_from_link}
318 self.comes_from = comes_from
319 self.requires_python = requires_python if requires_python else None
320 self.yanked_reason = yanked_reason
321 self.metadata_file_data = metadata_file_data
322 self.upload_time = upload_time
324 self.cache_link_parsing = cache_link_parsing
325 self.egg_fragment = self._egg_fragment()
327 @classmethod
328 def from_json(
329 cls,
330 file_data: dict[str, Any],
331 page_url: str,
332 ) -> Link | None:
333 """
334 Convert an pypi json document from a simple repository page into a Link.
335 """
336 file_url = file_data.get("url")
337 if file_url is None:
338 return None
340 url = _ensure_quoted_url(_absolute_link_url(page_url, file_url))
341 pyrequire = file_data.get("requires-python")
342 yanked_reason = file_data.get("yanked")
343 hashes = file_data.get("hashes", {})
345 # PEP 714: Indexes must use the name core-metadata, but
346 # clients should support the old name as a fallback for compatibility.
347 metadata_info = file_data.get("core-metadata")
348 if metadata_info is None:
349 metadata_info = file_data.get("dist-info-metadata")
351 if upload_time_data := file_data.get("upload-time"):
352 upload_time = parse_iso_datetime(upload_time_data)
353 else:
354 upload_time = None
356 # The metadata info value may be a boolean, or a dict of hashes.
357 if isinstance(metadata_info, dict):
358 # The file exists, and hashes have been supplied
359 metadata_file_data = MetadataFile(supported_hashes(metadata_info))
360 elif metadata_info:
361 # The file exists, but there are no hashes
362 metadata_file_data = MetadataFile(None)
363 else:
364 # False or not present: the file does not exist
365 metadata_file_data = None
367 # The Link.yanked_reason expects an empty string instead of a boolean.
368 if yanked_reason and not isinstance(yanked_reason, str):
369 yanked_reason = ""
370 # The Link.yanked_reason expects None instead of False.
371 elif not yanked_reason:
372 yanked_reason = None
374 return cls(
375 url,
376 comes_from=page_url,
377 requires_python=pyrequire,
378 yanked_reason=yanked_reason,
379 hashes=hashes,
380 metadata_file_data=metadata_file_data,
381 upload_time=upload_time,
382 )
384 @classmethod
385 def from_element(
386 cls,
387 anchor_attribs: dict[str, str | None],
388 page_url: str,
389 base_url: str,
390 ) -> Link | None:
391 """
392 Convert an anchor element's attributes in a simple repository page to a Link.
393 """
394 href = anchor_attribs.get("href")
395 if not href:
396 return None
398 url = _ensure_quoted_url(_absolute_link_url(base_url, href))
399 pyrequire = anchor_attribs.get("data-requires-python")
400 yanked_reason = anchor_attribs.get("data-yanked")
402 # PEP 714: Indexes must use the name data-core-metadata, but
403 # clients should support the old name as a fallback for compatibility.
404 metadata_info = anchor_attribs.get("data-core-metadata")
405 if metadata_info is None:
406 metadata_info = anchor_attribs.get("data-dist-info-metadata")
407 # The metadata info value may be the string "true", or a string of
408 # the form "hashname=hashval"
409 if metadata_info == "true":
410 # The file exists, but there are no hashes
411 metadata_file_data = MetadataFile(None)
412 elif metadata_info is None:
413 # The file does not exist
414 metadata_file_data = None
415 else:
416 # The file exists, and hashes have been supplied
417 hashname, sep, hashval = metadata_info.partition("=")
418 if sep == "=":
419 metadata_file_data = MetadataFile(supported_hashes({hashname: hashval}))
420 else:
421 # Error - data is wrong. Treat as no hashes supplied.
422 logger.debug(
423 "Index returned invalid data-dist-info-metadata value: %s",
424 metadata_info,
425 )
426 metadata_file_data = MetadataFile(None)
428 return cls(
429 url,
430 comes_from=page_url,
431 requires_python=pyrequire,
432 yanked_reason=yanked_reason,
433 metadata_file_data=metadata_file_data,
434 )
436 def __str__(self) -> str:
437 if self.requires_python:
438 rp = f" (requires-python:{self.requires_python})"
439 else:
440 rp = ""
441 if self.comes_from:
442 return f"{self.redacted_url} (from {self.comes_from}){rp}"
443 else:
444 return self.redacted_url
446 def __repr__(self) -> str:
447 return f"<Link {self}>"
449 def __hash__(self) -> int:
450 return hash(self.url)
452 def __eq__(self, other: Any) -> bool:
453 if not isinstance(other, Link):
454 return NotImplemented
455 return self.url == other.url
457 def __lt__(self, other: Any) -> bool:
458 if not isinstance(other, Link):
459 return NotImplemented
460 return self.url < other.url
462 @property
463 def url(self) -> str:
464 return self._url
466 @property
467 def redacted_url(self) -> str:
468 return redact_auth_from_url(self.url)
470 @property
471 def filename(self) -> PathComponent:
472 name = _to_path_component(posixpath.basename(self.path.rstrip("/")))
473 if name:
474 return name
476 # No component in the path; fall back to the netloc, dropping any auth.
477 return _to_path_component(split_auth_from_netloc(self.netloc)[0])
479 @property
480 def file_path(self) -> str:
481 return url_to_path(self.url)
483 @property
484 def scheme(self) -> str:
485 return self._parsed_url.scheme
487 @property
488 def netloc(self) -> str:
489 """
490 This can contain auth information.
491 """
492 return self._parsed_url.netloc
494 @property
495 def path(self) -> str:
496 return self._path
498 def splitext(self) -> tuple[str, str]:
499 return splitext(posixpath.basename(self.path.rstrip("/")))
501 @property
502 def ext(self) -> str:
503 return self.splitext()[1]
505 @property
506 def url_without_fragment(self) -> str:
507 scheme, netloc, path, query, fragment = self._parsed_url
508 return urllib.parse.urlunsplit((scheme, netloc, path, query, ""))
510 _egg_fragment_re = re.compile(r"[#&]egg=([^&]*)")
512 # Per PEP 508.
513 _project_name_re = re.compile(
514 r"^([A-Z0-9]|[A-Z0-9][A-Z0-9._-]*[A-Z0-9])$", re.IGNORECASE
515 )
517 def _egg_fragment(self) -> str | None:
518 match = self._egg_fragment_re.search(self._url)
519 if not match:
520 return None
522 # An egg fragment looks like a PEP 508 project name, along with
523 # an optional extras specifier. Anything else is invalid.
524 project_name = match.group(1)
525 if not self._project_name_re.match(project_name):
526 raise InvalidEggFragment(self, project_name)
528 return project_name
530 _subdirectory_fragment_re = re.compile(r"[#&]subdirectory=([^&]*)")
532 @property
533 def subdirectory_fragment(self) -> str | None:
534 match = self._subdirectory_fragment_re.search(self._url)
535 if not match:
536 return None
537 return match.group(1)
539 def metadata_link(self) -> Link | None:
540 """Return a link to the associated core metadata file (if any)."""
541 if self.metadata_file_data is None:
542 return None
543 metadata_url = f"{self.url_without_fragment}.metadata"
544 if self.metadata_file_data.hashes is None:
545 return Link(metadata_url)
546 return Link(metadata_url, hashes=self.metadata_file_data.hashes)
548 def as_hashes(self) -> Hashes:
549 return Hashes({k: [v] for k, v in self._hashes.items()})
551 @property
552 def hash(self) -> str | None:
553 return next(iter(self._hashes.values()), None)
555 @property
556 def hash_name(self) -> str | None:
557 return next(iter(self._hashes), None)
559 @property
560 def show_url(self) -> str:
561 return posixpath.basename(self._url.split("#", 1)[0].split("?", 1)[0])
563 @property
564 def is_file(self) -> bool:
565 return self.scheme == "file"
567 def is_existing_dir(self) -> bool:
568 return self.is_file and os.path.isdir(self.file_path)
570 @property
571 def is_wheel(self) -> bool:
572 return self.ext == WHEEL_EXTENSION
574 @property
575 def is_vcs(self) -> bool:
576 from pip._internal.vcs import vcs
578 return self.scheme in vcs.all_schemes
580 @property
581 def is_yanked(self) -> bool:
582 return self.yanked_reason is not None
584 @property
585 def has_hash(self) -> bool:
586 return bool(self._hashes)
588 def is_hash_allowed(self, hashes: Hashes | None) -> bool:
589 """
590 Return True if the link has a hash and it is allowed by `hashes`.
591 """
592 if hashes is None:
593 return False
594 return any(hashes.is_hash_allowed(k, v) for k, v in self._hashes.items())
597class _CleanResult(NamedTuple):
598 """Convert link for equivalency check.
600 This is used in the resolver to check whether two URL-specified requirements
601 likely point to the same distribution and can be considered equivalent. This
602 equivalency logic avoids comparing URLs literally, which can be too strict
603 (e.g. "a=1&b=2" vs "b=2&a=1") and produce conflicts unexpecting to users.
605 Currently this does three things:
607 1. Drop the basic auth part. This is technically wrong since a server can
608 serve different content based on auth, but if it does that, it is even
609 impossible to guarantee two URLs without auth are equivalent, since
610 the user can input different auth information when prompted. So the
611 practical solution is to assume the auth doesn't affect the response.
612 2. Parse the query to avoid the ordering issue. Note that ordering under the
613 same key in the query are NOT cleaned; i.e. "a=1&a=2" and "a=2&a=1" are
614 still considered different.
615 3. Explicitly drop most of the fragment part, except ``subdirectory=`` and
616 hash values, since it should have no impact the downloaded content. Note
617 that this drops the "egg=" part historically used to denote the requested
618 project (and extras), which is wrong in the strictest sense, but too many
619 people are supplying it inconsistently to cause superfluous resolution
620 conflicts, so we choose to also ignore them.
621 """
623 parsed: urllib.parse.SplitResult
624 query: dict[str, list[str]]
625 subdirectory: str
626 hashes: dict[str, str]
629def _clean_link(link: Link) -> _CleanResult:
630 parsed = link._parsed_url
631 netloc = parsed.netloc.rsplit("@", 1)[-1]
632 # According to RFC 8089, an empty host in file: means localhost.
633 if parsed.scheme == "file" and not netloc:
634 netloc = "localhost"
635 fragment = urllib.parse.parse_qs(parsed.fragment)
636 if "egg" in fragment:
637 logger.debug("Ignoring egg= fragment in %s", link)
638 try:
639 # If there are multiple subdirectory values, use the first one.
640 # This matches the behavior of Link.subdirectory_fragment.
641 subdirectory = fragment["subdirectory"][0]
642 except (IndexError, KeyError):
643 subdirectory = ""
644 # If there are multiple hash values under the same algorithm, use the
645 # first one. This matches the behavior of Link.hash_value.
646 hashes = {k: fragment[k][0] for k in _SUPPORTED_HASHES if k in fragment}
647 return _CleanResult(
648 parsed=parsed._replace(netloc=netloc, query="", fragment=""),
649 query=urllib.parse.parse_qs(parsed.query),
650 subdirectory=subdirectory,
651 hashes=hashes,
652 )
655@functools.cache
656def links_equivalent(link1: Link, link2: Link) -> bool:
657 return _clean_link(link1) == _clean_link(link2)