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
11import urllib.request
12from collections.abc import Mapping
13from dataclasses import dataclass
14from typing import (
15 Any,
16 NamedTuple,
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# Order matters, earlier hashes have a precedence over later hashes for what
35# we will pick to use.
36_SUPPORTED_HASHES = ("sha512", "sha384", "sha256", "sha224", "sha1", "md5")
39@dataclass(frozen=True)
40class LinkHash:
41 """Links to content may have embedded hash values. This class parses those.
43 `name` must be any member of `_SUPPORTED_HASHES`.
45 This class can be converted to and from `ArchiveInfo`. While ArchiveInfo intends to
46 be JSON-serializable to conform to PEP 610, this class contains the logic for
47 parsing a hash name and value for correctness, and then checking whether that hash
48 conforms to a schema with `.is_hash_allowed()`."""
50 name: str
51 value: str
53 _hash_url_fragment_re = re.compile(
54 # NB: we do not validate that the second group (.*) is a valid hex
55 # digest. Instead, we simply keep that string in this class, and then check it
56 # against Hashes when hash-checking is needed. This is easier to debug than
57 # proactively discarding an invalid hex digest, as we handle incorrect hashes
58 # and malformed hashes in the same place.
59 r"[#&]({choices})=([^&]*)".format(
60 choices="|".join(re.escape(hash_name) for hash_name in _SUPPORTED_HASHES)
61 ),
62 )
64 def __post_init__(self) -> None:
65 assert self.name in _SUPPORTED_HASHES
67 @classmethod
68 @functools.cache
69 def find_hash_url_fragment(cls, url: str) -> LinkHash | None:
70 """Search a string for a checksum algorithm name and encoded output value."""
71 match = cls._hash_url_fragment_re.search(url)
72 if match is None:
73 return None
74 name, value = match.groups()
75 return cls(name=name, value=value)
77 def as_dict(self) -> dict[str, str]:
78 return {self.name: self.value}
80 def as_hashes(self) -> Hashes:
81 """Return a Hashes instance which checks only for the current hash."""
82 return Hashes({self.name: [self.value]})
84 def is_hash_allowed(self, hashes: Hashes | None) -> bool:
85 """
86 Return True if the current hash is allowed by `hashes`.
87 """
88 if hashes is None:
89 return False
90 return hashes.is_hash_allowed(self.name, hex_digest=self.value)
93@dataclass(frozen=True)
94class MetadataFile:
95 """Information about a core metadata file associated with a distribution."""
97 hashes: dict[str, str] | None
99 def __post_init__(self) -> None:
100 if self.hashes is not None:
101 assert all(name in _SUPPORTED_HASHES for name in self.hashes)
104def supported_hashes(hashes: dict[str, str] | None) -> dict[str, str] | None:
105 # Remove any unsupported hash types from the mapping. If this leaves no
106 # supported hashes, return None
107 if hashes is None:
108 return None
109 hashes = {n: v for n, v in hashes.items() if n in _SUPPORTED_HASHES}
110 if not hashes:
111 return None
112 return hashes
115def _clean_url_path_part(part: str) -> str:
116 """
117 Clean a "part" of a URL path (i.e. after splitting on "@" characters).
118 """
119 # We unquote prior to quoting to make sure nothing is double quoted.
120 return urllib.parse.quote(urllib.parse.unquote(part))
123def _clean_file_url_path(part: str) -> str:
124 """
125 Clean the first part of a URL path that corresponds to a local
126 filesystem path (i.e. the first part after splitting on "@" characters).
127 """
128 # We unquote prior to quoting to make sure nothing is double quoted.
129 # Also, on Windows the path part might contain a drive letter which
130 # should not be quoted. On Linux where drive letters do not
131 # exist, the colon should be quoted. We rely on urllib.request
132 # to do the right thing here.
133 ret = urllib.request.pathname2url(urllib.request.url2pathname(part))
134 if ret.startswith("///"):
135 # Remove any URL authority section, leaving only the URL path.
136 ret = ret.removeprefix("//")
137 return ret
140# percent-encoded: /
141_reserved_chars_re = re.compile("(@|%2F)", re.IGNORECASE)
144def _clean_url_path(path: str, is_local_path: bool) -> str:
145 """
146 Clean the path portion of a URL.
147 """
148 if is_local_path:
149 clean_func = _clean_file_url_path
150 else:
151 clean_func = _clean_url_path_part
153 # Split on the reserved characters prior to cleaning so that
154 # revision strings in VCS URLs are properly preserved.
155 parts = _reserved_chars_re.split(path)
157 cleaned_parts = []
158 for to_clean, reserved in pairwise(itertools.chain(parts, [""])):
159 cleaned_parts.append(clean_func(to_clean))
160 # Normalize %xx escapes (e.g. %2f -> %2F)
161 cleaned_parts.append(reserved.upper())
163 return "".join(cleaned_parts)
166def _ensure_quoted_url(url: str) -> str:
167 """
168 Make sure a link is fully quoted.
169 For example, if ' ' occurs in the URL, it will be replaced with "%20",
170 and without double-quoting other characters.
171 """
172 # Split the URL into parts according to the general structure
173 # `scheme://netloc/path?query#fragment`.
174 result = urllib.parse.urlsplit(url)
175 # If the netloc is empty, then the URL refers to a local filesystem path.
176 is_local_path = not result.netloc
177 path = _clean_url_path(result.path, is_local_path=is_local_path)
178 # Temporarily replace scheme with file to ensure the URL generated by
179 # urlunsplit() contains an empty netloc (file://) as per RFC 1738.
180 ret = urllib.parse.urlunsplit(result._replace(scheme="file", path=path))
181 ret = result.scheme + ret[4:] # Restore original scheme.
182 return ret
185def _absolute_link_url(base_url: str, url: str) -> str:
186 """
187 A faster implementation of urllib.parse.urljoin with a shortcut
188 for absolute http/https URLs.
189 """
190 if url.startswith(("https://", "http://")):
191 return url
192 else:
193 return urllib.parse.urljoin(base_url, url)
196@functools.total_ordering
197class Link:
198 """Represents a parsed link from a Package Index's simple URL"""
200 __slots__ = [
201 "_parsed_url",
202 "_url",
203 "_path",
204 "_hashes",
205 "comes_from",
206 "requires_python",
207 "yanked_reason",
208 "metadata_file_data",
209 "upload_time",
210 "cache_link_parsing",
211 "egg_fragment",
212 ]
214 def __init__(
215 self,
216 url: str,
217 comes_from: str | None = None,
218 requires_python: str | None = None,
219 yanked_reason: str | None = None,
220 metadata_file_data: MetadataFile | None = None,
221 upload_time: datetime.datetime | None = None,
222 cache_link_parsing: bool = True,
223 hashes: Mapping[str, str] | None = None,
224 ) -> None:
225 """
226 :param url: url of the resource pointed to (href of the link)
227 :param comes_from: URL or string indicating where the link was found.
228 :param requires_python: String containing the `Requires-Python`
229 metadata field, specified in PEP 345. This may be specified by
230 a data-requires-python attribute in the HTML link tag, as
231 described in PEP 503.
232 :param yanked_reason: the reason the file has been yanked, if the
233 file has been yanked, or None if the file hasn't been yanked.
234 This is the value of the "data-yanked" attribute, if present, in
235 a simple repository HTML link. If the file has been yanked but
236 no reason was provided, this should be the empty string. See
237 PEP 592 for more information and the specification.
238 :param metadata_file_data: the metadata attached to the file, or None if
239 no such metadata is provided. This argument, if not None, indicates
240 that a separate metadata file exists, and also optionally supplies
241 hashes for that file.
242 :param upload_time: upload time of the file, or None if the information
243 is not available from the server.
244 :param cache_link_parsing: A flag that is used elsewhere to determine
245 whether resources retrieved from this link should be cached. PyPI
246 URLs should generally have this set to False, for example.
247 :param hashes: A mapping of hash names to digests to allow us to
248 determine the validity of a download.
249 """
251 # The comes_from, requires_python, and metadata_file_data arguments are
252 # only used by classmethods of this class, and are not used in client
253 # code directly.
255 # url can be a UNC windows share
256 if url.startswith("\\\\"):
257 url = path_to_url(url)
259 self._parsed_url = urllib.parse.urlsplit(url)
260 # Store the url as a private attribute to prevent accidentally
261 # trying to set a new value.
262 self._url = url
263 # The .path property is hot, so calculate its value ahead of time.
264 self._path = urllib.parse.unquote(self._parsed_url.path)
266 link_hash = LinkHash.find_hash_url_fragment(url)
267 hashes_from_link = {} if link_hash is None else link_hash.as_dict()
268 if hashes is None:
269 self._hashes = hashes_from_link
270 else:
271 self._hashes = {**hashes, **hashes_from_link}
273 self.comes_from = comes_from
274 self.requires_python = requires_python if requires_python else None
275 self.yanked_reason = yanked_reason
276 self.metadata_file_data = metadata_file_data
277 self.upload_time = upload_time
279 self.cache_link_parsing = cache_link_parsing
280 self.egg_fragment = self._egg_fragment()
282 @classmethod
283 def from_json(
284 cls,
285 file_data: dict[str, Any],
286 page_url: str,
287 ) -> Link | None:
288 """
289 Convert an pypi json document from a simple repository page into a Link.
290 """
291 file_url = file_data.get("url")
292 if file_url is None:
293 return None
295 url = _ensure_quoted_url(_absolute_link_url(page_url, file_url))
296 pyrequire = file_data.get("requires-python")
297 yanked_reason = file_data.get("yanked")
298 hashes = file_data.get("hashes", {})
300 # PEP 714: Indexes must use the name core-metadata, but
301 # clients should support the old name as a fallback for compatibility.
302 metadata_info = file_data.get("core-metadata")
303 if metadata_info is None:
304 metadata_info = file_data.get("dist-info-metadata")
306 if upload_time_data := file_data.get("upload-time"):
307 upload_time = parse_iso_datetime(upload_time_data)
308 else:
309 upload_time = None
311 # The metadata info value may be a boolean, or a dict of hashes.
312 if isinstance(metadata_info, dict):
313 # The file exists, and hashes have been supplied
314 metadata_file_data = MetadataFile(supported_hashes(metadata_info))
315 elif metadata_info:
316 # The file exists, but there are no hashes
317 metadata_file_data = MetadataFile(None)
318 else:
319 # False or not present: the file does not exist
320 metadata_file_data = None
322 # The Link.yanked_reason expects an empty string instead of a boolean.
323 if yanked_reason and not isinstance(yanked_reason, str):
324 yanked_reason = ""
325 # The Link.yanked_reason expects None instead of False.
326 elif not yanked_reason:
327 yanked_reason = None
329 return cls(
330 url,
331 comes_from=page_url,
332 requires_python=pyrequire,
333 yanked_reason=yanked_reason,
334 hashes=hashes,
335 metadata_file_data=metadata_file_data,
336 upload_time=upload_time,
337 )
339 @classmethod
340 def from_element(
341 cls,
342 anchor_attribs: dict[str, str | None],
343 page_url: str,
344 base_url: str,
345 ) -> Link | None:
346 """
347 Convert an anchor element's attributes in a simple repository page to a Link.
348 """
349 href = anchor_attribs.get("href")
350 if not href:
351 return None
353 url = _ensure_quoted_url(_absolute_link_url(base_url, href))
354 pyrequire = anchor_attribs.get("data-requires-python")
355 yanked_reason = anchor_attribs.get("data-yanked")
357 # PEP 714: Indexes must use the name data-core-metadata, but
358 # clients should support the old name as a fallback for compatibility.
359 metadata_info = anchor_attribs.get("data-core-metadata")
360 if metadata_info is None:
361 metadata_info = anchor_attribs.get("data-dist-info-metadata")
362 # The metadata info value may be the string "true", or a string of
363 # the form "hashname=hashval"
364 if metadata_info == "true":
365 # The file exists, but there are no hashes
366 metadata_file_data = MetadataFile(None)
367 elif metadata_info is None:
368 # The file does not exist
369 metadata_file_data = None
370 else:
371 # The file exists, and hashes have been supplied
372 hashname, sep, hashval = metadata_info.partition("=")
373 if sep == "=":
374 metadata_file_data = MetadataFile(supported_hashes({hashname: hashval}))
375 else:
376 # Error - data is wrong. Treat as no hashes supplied.
377 logger.debug(
378 "Index returned invalid data-dist-info-metadata value: %s",
379 metadata_info,
380 )
381 metadata_file_data = MetadataFile(None)
383 return cls(
384 url,
385 comes_from=page_url,
386 requires_python=pyrequire,
387 yanked_reason=yanked_reason,
388 metadata_file_data=metadata_file_data,
389 )
391 def __str__(self) -> str:
392 if self.requires_python:
393 rp = f" (requires-python:{self.requires_python})"
394 else:
395 rp = ""
396 if self.comes_from:
397 return f"{self.redacted_url} (from {self.comes_from}){rp}"
398 else:
399 return self.redacted_url
401 def __repr__(self) -> str:
402 return f"<Link {self}>"
404 def __hash__(self) -> int:
405 return hash(self.url)
407 def __eq__(self, other: Any) -> bool:
408 if not isinstance(other, Link):
409 return NotImplemented
410 return self.url == other.url
412 def __lt__(self, other: Any) -> bool:
413 if not isinstance(other, Link):
414 return NotImplemented
415 return self.url < other.url
417 @property
418 def url(self) -> str:
419 return self._url
421 @property
422 def redacted_url(self) -> str:
423 return redact_auth_from_url(self.url)
425 @property
426 def filename(self) -> str:
427 path = self.path.rstrip("/")
428 name = posixpath.basename(path)
429 if not name:
430 # Make sure we don't leak auth information if the netloc
431 # includes a username and password.
432 netloc, user_pass = split_auth_from_netloc(self.netloc)
433 return netloc
435 name = urllib.parse.unquote(name)
436 assert name, f"URL {self._url!r} produced no filename"
437 return name
439 @property
440 def file_path(self) -> str:
441 return url_to_path(self.url)
443 @property
444 def scheme(self) -> str:
445 return self._parsed_url.scheme
447 @property
448 def netloc(self) -> str:
449 """
450 This can contain auth information.
451 """
452 return self._parsed_url.netloc
454 @property
455 def path(self) -> str:
456 return self._path
458 def splitext(self) -> tuple[str, str]:
459 return splitext(posixpath.basename(self.path.rstrip("/")))
461 @property
462 def ext(self) -> str:
463 return self.splitext()[1]
465 @property
466 def url_without_fragment(self) -> str:
467 scheme, netloc, path, query, fragment = self._parsed_url
468 return urllib.parse.urlunsplit((scheme, netloc, path, query, ""))
470 _egg_fragment_re = re.compile(r"[#&]egg=([^&]*)")
472 # Per PEP 508.
473 _project_name_re = re.compile(
474 r"^([A-Z0-9]|[A-Z0-9][A-Z0-9._-]*[A-Z0-9])$", re.IGNORECASE
475 )
477 def _egg_fragment(self) -> str | None:
478 match = self._egg_fragment_re.search(self._url)
479 if not match:
480 return None
482 # An egg fragment looks like a PEP 508 project name, along with
483 # an optional extras specifier. Anything else is invalid.
484 project_name = match.group(1)
485 if not self._project_name_re.match(project_name):
486 raise InvalidEggFragment(self, project_name)
488 return project_name
490 _subdirectory_fragment_re = re.compile(r"[#&]subdirectory=([^&]*)")
492 @property
493 def subdirectory_fragment(self) -> str | None:
494 match = self._subdirectory_fragment_re.search(self._url)
495 if not match:
496 return None
497 return match.group(1)
499 def metadata_link(self) -> Link | None:
500 """Return a link to the associated core metadata file (if any)."""
501 if self.metadata_file_data is None:
502 return None
503 metadata_url = f"{self.url_without_fragment}.metadata"
504 if self.metadata_file_data.hashes is None:
505 return Link(metadata_url)
506 return Link(metadata_url, hashes=self.metadata_file_data.hashes)
508 def as_hashes(self) -> Hashes:
509 return Hashes({k: [v] for k, v in self._hashes.items()})
511 @property
512 def hash(self) -> str | None:
513 return next(iter(self._hashes.values()), None)
515 @property
516 def hash_name(self) -> str | None:
517 return next(iter(self._hashes), None)
519 @property
520 def show_url(self) -> str:
521 return posixpath.basename(self._url.split("#", 1)[0].split("?", 1)[0])
523 @property
524 def is_file(self) -> bool:
525 return self.scheme == "file"
527 def is_existing_dir(self) -> bool:
528 return self.is_file and os.path.isdir(self.file_path)
530 @property
531 def is_wheel(self) -> bool:
532 return self.ext == WHEEL_EXTENSION
534 @property
535 def is_vcs(self) -> bool:
536 from pip._internal.vcs import vcs
538 return self.scheme in vcs.all_schemes
540 @property
541 def is_yanked(self) -> bool:
542 return self.yanked_reason is not None
544 @property
545 def has_hash(self) -> bool:
546 return bool(self._hashes)
548 def is_hash_allowed(self, hashes: Hashes | None) -> bool:
549 """
550 Return True if the link has a hash and it is allowed by `hashes`.
551 """
552 if hashes is None:
553 return False
554 return any(hashes.is_hash_allowed(k, v) for k, v in self._hashes.items())
557class _CleanResult(NamedTuple):
558 """Convert link for equivalency check.
560 This is used in the resolver to check whether two URL-specified requirements
561 likely point to the same distribution and can be considered equivalent. This
562 equivalency logic avoids comparing URLs literally, which can be too strict
563 (e.g. "a=1&b=2" vs "b=2&a=1") and produce conflicts unexpecting to users.
565 Currently this does three things:
567 1. Drop the basic auth part. This is technically wrong since a server can
568 serve different content based on auth, but if it does that, it is even
569 impossible to guarantee two URLs without auth are equivalent, since
570 the user can input different auth information when prompted. So the
571 practical solution is to assume the auth doesn't affect the response.
572 2. Parse the query to avoid the ordering issue. Note that ordering under the
573 same key in the query are NOT cleaned; i.e. "a=1&a=2" and "a=2&a=1" are
574 still considered different.
575 3. Explicitly drop most of the fragment part, except ``subdirectory=`` and
576 hash values, since it should have no impact the downloaded content. Note
577 that this drops the "egg=" part historically used to denote the requested
578 project (and extras), which is wrong in the strictest sense, but too many
579 people are supplying it inconsistently to cause superfluous resolution
580 conflicts, so we choose to also ignore them.
581 """
583 parsed: urllib.parse.SplitResult
584 query: dict[str, list[str]]
585 subdirectory: str
586 hashes: dict[str, str]
589def _clean_link(link: Link) -> _CleanResult:
590 parsed = link._parsed_url
591 netloc = parsed.netloc.rsplit("@", 1)[-1]
592 # According to RFC 8089, an empty host in file: means localhost.
593 if parsed.scheme == "file" and not netloc:
594 netloc = "localhost"
595 fragment = urllib.parse.parse_qs(parsed.fragment)
596 if "egg" in fragment:
597 logger.debug("Ignoring egg= fragment in %s", link)
598 try:
599 # If there are multiple subdirectory values, use the first one.
600 # This matches the behavior of Link.subdirectory_fragment.
601 subdirectory = fragment["subdirectory"][0]
602 except (IndexError, KeyError):
603 subdirectory = ""
604 # If there are multiple hash values under the same algorithm, use the
605 # first one. This matches the behavior of Link.hash_value.
606 hashes = {k: fragment[k][0] for k in _SUPPORTED_HASHES if k in fragment}
607 return _CleanResult(
608 parsed=parsed._replace(netloc=netloc, query="", fragment=""),
609 query=urllib.parse.parse_qs(parsed.query),
610 subdirectory=subdirectory,
611 hashes=hashes,
612 )
615@functools.cache
616def links_equivalent(link1: Link, link2: Link) -> bool:
617 return _clean_link(link1) == _clean_link(link2)