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
1"""Utilities related archives."""
3from __future__ import annotations
5import logging
6import os
7import shutil
8import stat
9import sys
10import tarfile
11import zipfile
12from collections.abc import Iterable
13from zipfile import ZipInfo
15from pip._internal.exceptions import InstallationError
16from pip._internal.utils.filetypes import (
17 BZ2_EXTENSIONS,
18 TAR_EXTENSIONS,
19 XZ_EXTENSIONS,
20 ZIP_EXTENSIONS,
21)
22from pip._internal.utils.misc import ensure_dir
24logger = logging.getLogger(__name__)
27SUPPORTED_EXTENSIONS = ZIP_EXTENSIONS + TAR_EXTENSIONS
29try:
30 import bz2 # noqa
32 SUPPORTED_EXTENSIONS += BZ2_EXTENSIONS
33except ImportError:
34 logger.debug("bz2 module is not available")
36try:
37 # Only for Python 3.3+
38 import lzma # noqa
40 SUPPORTED_EXTENSIONS += XZ_EXTENSIONS
41except ImportError:
42 logger.debug("lzma module is not available")
45def current_umask() -> int:
46 """Get the current umask which involves having to set it temporarily."""
47 mask = os.umask(0)
48 os.umask(mask)
49 return mask
52def split_leading_dir(path: str) -> list[str]:
53 path = path.lstrip("/").lstrip("\\")
54 if "/" in path and (
55 ("\\" in path and path.find("/") < path.find("\\")) or "\\" not in path
56 ):
57 return path.split("/", 1)
58 elif "\\" in path:
59 return path.split("\\", 1)
60 else:
61 return [path, ""]
64def has_leading_dir(paths: Iterable[str]) -> bool:
65 """Returns true if all the paths have the same leading path name
66 (i.e., everything is in one subdirectory in an archive)"""
67 common_prefix = None
68 for path in paths:
69 prefix, rest = split_leading_dir(path)
70 if not prefix:
71 return False
72 elif common_prefix is None:
73 common_prefix = prefix
74 elif prefix != common_prefix:
75 return False
76 return True
79def is_within_directory(
80 directory: str, target: str, *, resolve_symlinks: bool = False
81) -> bool:
82 """
83 Return true if the absolute path of target is within the directory
84 (including when target is equal to the directory).
86 When ``resolve_symlinks`` is true, resolve symlinks before comparing so
87 traversal through a symlink (e.g. "link/../file") is also caught.
88 """
89 if resolve_symlinks:
90 abs_directory = os.path.realpath(directory)
91 abs_target = os.path.realpath(target)
92 else:
93 abs_directory = os.path.abspath(directory)
94 abs_target = os.path.abspath(target)
96 return abs_target == abs_directory or abs_target.startswith(abs_directory + os.sep)
99def _get_default_mode_plus_executable() -> int:
100 return 0o777 & ~current_umask() | 0o111
103def set_extracted_file_to_default_mode_plus_executable(path: str) -> None:
104 """
105 Make file present at path have execute for user/group/world
106 (chmod +x) is no-op on windows per python docs
107 """
108 os.chmod(path, _get_default_mode_plus_executable())
111def zip_item_is_executable(info: ZipInfo) -> bool:
112 mode = info.external_attr >> 16
113 # if mode and regular file and any execute permissions for
114 # user/group/world?
115 return bool(mode and stat.S_ISREG(mode) and mode & 0o111)
118def unzip_file(filename: str, location: str, flatten: bool = True) -> None:
119 """
120 Unzip the file (with path `filename`) to the destination `location`. All
121 files are written based on system defaults and umask (i.e. permissions are
122 not preserved), except that regular file members with any execute
123 permissions (user, group, or world) have "chmod +x" applied after being
124 written. Note that for windows, any execute changes using os.chmod are
125 no-ops per the python docs.
126 """
127 ensure_dir(location)
128 zipfp = open(filename, "rb")
129 try:
130 zip = zipfile.ZipFile(zipfp, allowZip64=True)
131 leading = has_leading_dir(zip.namelist()) and flatten
132 for info in zip.infolist():
133 name = info.filename
134 fn = name
135 if leading:
136 fn = split_leading_dir(name)[1]
137 fn = os.path.join(location, fn)
138 dir = os.path.dirname(fn)
139 if not is_within_directory(location, fn):
140 message = (
141 "The zip file ({}) has a file ({}) trying to install "
142 "outside target directory ({})"
143 )
144 raise InstallationError(message.format(filename, fn, location))
145 if fn.endswith(("/", "\\")):
146 # A directory
147 ensure_dir(fn)
148 else:
149 ensure_dir(dir)
150 # Don't use read() to avoid allocating an arbitrarily large
151 # chunk of memory for the file's content
152 fp = zip.open(name)
153 try:
154 with open(fn, "wb") as destfp:
155 shutil.copyfileobj(fp, destfp)
156 finally:
157 fp.close()
158 if zip_item_is_executable(info):
159 set_extracted_file_to_default_mode_plus_executable(fn)
160 finally:
161 zipfp.close()
164def untar_file(filename: str, location: str) -> None:
165 """
166 Untar the file (with path `filename`) to the destination `location`.
167 All files are written based on system defaults and umask (i.e. permissions
168 are not preserved), except that regular file members with any execute
169 permissions (user, group, or world) have "chmod +x" applied on top of the
170 default. Note that for windows, any execute changes using os.chmod are
171 no-ops per the python docs.
172 """
173 ensure_dir(location)
174 if filename.lower().endswith(".gz") or filename.lower().endswith(".tgz"):
175 mode = "r:gz"
176 elif filename.lower().endswith(BZ2_EXTENSIONS):
177 mode = "r:bz2"
178 elif filename.lower().endswith(XZ_EXTENSIONS):
179 mode = "r:xz"
180 elif filename.lower().endswith(".tar"):
181 mode = "r"
182 else:
183 logger.warning(
184 "Cannot determine compression type for file %s",
185 filename,
186 )
187 mode = "r:*"
189 tar = tarfile.open(filename, mode, encoding="utf-8") # type: ignore
190 try:
191 leading = has_leading_dir([member.name for member in tar.getmembers()])
193 # PEP 706 added `tarfile.data_filter`, and made some other changes to
194 # Python's tarfile module (see below). The features were backported to
195 # security releases.
196 try:
197 data_filter = tarfile.data_filter
198 except AttributeError:
199 _untar_without_filter(filename, location, tar, leading)
200 else:
201 default_mode_plus_executable = _get_default_mode_plus_executable()
203 if leading:
204 # Strip the leading directory from all files in the archive,
205 # including hardlink targets (which are relative to the
206 # unpack location).
207 for member in tar.getmembers():
208 name_lead, name_rest = split_leading_dir(member.name)
209 member.name = name_rest
210 if member.islnk():
211 lnk_lead, lnk_rest = split_leading_dir(member.linkname)
212 if lnk_lead == name_lead:
213 member.linkname = lnk_rest
215 def pip_filter(member: tarfile.TarInfo, path: str) -> tarfile.TarInfo:
216 orig_mode = member.mode
217 try:
218 try:
219 member = data_filter(member, location)
220 except tarfile.LinkOutsideDestinationError:
221 if sys.version_info[:3] in {
222 (3, 9, 17),
223 (3, 10, 12),
224 (3, 11, 4),
225 }:
226 # The tarfile filter in specific Python versions
227 # raises LinkOutsideDestinationError on valid input
228 # (https://github.com/python/cpython/issues/107845)
229 # Ignore the error there, but do use the
230 # more lax `tar_filter`
231 member = tarfile.tar_filter(member, location)
232 else:
233 raise
234 except tarfile.TarError as exc:
235 message = "Invalid member in the tar file {}: {}"
236 # Filter error messages mention the member name.
237 # No need to add it here.
238 raise InstallationError(
239 message.format(
240 filename,
241 exc,
242 )
243 )
244 if member.isfile() and orig_mode & 0o111:
245 member.mode = default_mode_plus_executable
246 else:
247 # See PEP 706 note above.
248 # The PEP changed this from `int` to `Optional[int]`,
249 # where None means "use the default". Mypy doesn't
250 # know this yet.
251 member.mode = None # type: ignore [assignment]
252 return member
254 tar.extractall(location, filter=pip_filter)
256 finally:
257 tar.close()
260def is_symlink_target_in_tar(tar: tarfile.TarFile, tarinfo: tarfile.TarInfo) -> bool:
261 """Check if the file pointed to by the symbolic link is in the tar archive"""
262 linkname = os.path.join(os.path.dirname(tarinfo.name), tarinfo.linkname)
264 linkname = os.path.normpath(linkname)
265 linkname = linkname.replace("\\", "/")
267 try:
268 tar.getmember(linkname)
269 return True
270 except KeyError:
271 return False
274def _untar_without_filter(
275 filename: str,
276 location: str,
277 tar: tarfile.TarFile,
278 leading: bool,
279) -> None:
280 """Fallback for Python without tarfile.data_filter"""
281 # NOTE: This function can be removed once pip requires CPython ≥ 3.12.
282 # PEP 706 added tarfile.data_filter, made tarfile extraction operations more secure.
283 # This feature is fully supported from CPython 3.12 onward.
284 for member in tar.getmembers():
285 fn = member.name
286 if leading:
287 fn = split_leading_dir(fn)[1]
288 path = os.path.join(location, fn)
290 # The plain check rejects textual ".." escapes; resolving symlinks also
291 # catches a later member redirected outside by an earlier member's
292 # symlink (e.g. "link/../file").
293 if not is_within_directory(location, path) or not is_within_directory(
294 location, path, resolve_symlinks=True
295 ):
296 message = (
297 "The tar file ({}) has a file ({}) trying to install "
298 "outside target directory ({})"
299 )
300 raise InstallationError(message.format(filename, path, location))
301 if member.isdir():
302 ensure_dir(path)
303 elif member.issym():
304 # Reject symlinks resolving outside the destination, so a later
305 # member cannot be written through them.
306 target = os.path.join(os.path.dirname(path), member.linkname)
307 if not is_within_directory(location, target, resolve_symlinks=True):
308 message = (
309 "The tar file ({}) has a file ({}) trying to install "
310 "outside target directory ({})"
311 )
312 raise InstallationError(
313 message.format(filename, member.name, member.linkname)
314 )
315 if not is_symlink_target_in_tar(tar, member):
316 message = (
317 "The tar file ({}) has a file ({}) trying to install "
318 "outside target directory ({})"
319 )
320 raise InstallationError(
321 message.format(filename, member.name, member.linkname)
322 )
323 try:
324 tar._extract_member(member, path)
325 except Exception as exc:
326 # Some corrupt tar files seem to produce this
327 # (specifically bad symlinks)
328 logger.warning(
329 "In the tar file %s the member %s is invalid: %s",
330 filename,
331 member.name,
332 exc,
333 )
334 continue
335 else:
336 try:
337 fp = tar.extractfile(member)
338 except (KeyError, AttributeError) as exc:
339 # Some corrupt tar files seem to produce this
340 # (specifically bad symlinks)
341 logger.warning(
342 "In the tar file %s the member %s is invalid: %s",
343 filename,
344 member.name,
345 exc,
346 )
347 continue
348 ensure_dir(os.path.dirname(path))
349 assert fp is not None
350 with open(path, "wb") as destfp:
351 shutil.copyfileobj(fp, destfp)
352 fp.close()
353 # Update the timestamp (useful for cython compiled files)
354 tar.utime(member, path)
355 # member have any execute permissions for user/group/world?
356 if member.mode & 0o111:
357 set_extracted_file_to_default_mode_plus_executable(path)
360def unpack_file(
361 filename: str,
362 location: str,
363 content_type: str | None = None,
364) -> None:
365 """Unpack ``filename`` into ``location``.
367 Archive format is chosen in order of decreasing reliability:
368 ``content_type``, then filename extension, then magic signature
369 (unambiguous matches only).
370 """
371 filename = os.path.realpath(filename)
372 zip_flatten = not filename.endswith(".whl")
374 def _unzip() -> None:
375 unzip_file(filename, location, flatten=zip_flatten)
377 def _untar() -> None:
378 untar_file(filename, location)
380 if content_type == "application/zip":
381 return _unzip()
382 if content_type == "application/x-gzip":
383 return _untar()
385 if filename.lower().endswith(ZIP_EXTENSIONS):
386 return _unzip()
387 if filename.lower().endswith(TAR_EXTENSIONS + BZ2_EXTENSIONS + XZ_EXTENSIONS):
388 return _untar()
390 # avoid ambiguous case where both signature checks return True
391 is_zipfile = zipfile.is_zipfile(filename)
392 is_tarfile = tarfile.is_tarfile(filename)
393 if is_zipfile and not is_tarfile:
394 return _unzip()
395 if is_tarfile and not is_zipfile:
396 return _untar()
397 if is_zipfile and is_tarfile:
398 logger.error("Ambiguous file signature in %s.", filename)
400 logger.critical(
401 "Cannot unpack file %s (downloaded from %s, content-type: %s); "
402 "cannot detect archive format",
403 filename,
404 location,
405 content_type,
406 )
407 raise InstallationError(f"Cannot determine archive format of {location}")