unpacking.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267
  1. """Utilities related archives.
  2. """
  3. import logging
  4. import os
  5. import shutil
  6. import stat
  7. import tarfile
  8. import zipfile
  9. from typing import Iterable, List, Optional
  10. from zipfile import ZipInfo
  11. from pip._internal.exceptions import InstallationError
  12. from pip._internal.utils.filetypes import (
  13. BZ2_EXTENSIONS,
  14. TAR_EXTENSIONS,
  15. XZ_EXTENSIONS,
  16. ZIP_EXTENSIONS,
  17. )
  18. from pip._internal.utils.misc import ensure_dir
  19. logger = logging.getLogger(__name__)
  20. SUPPORTED_EXTENSIONS = ZIP_EXTENSIONS + TAR_EXTENSIONS
  21. try:
  22. import bz2 # noqa
  23. SUPPORTED_EXTENSIONS += BZ2_EXTENSIONS
  24. except ImportError:
  25. logger.debug("bz2 module is not available")
  26. try:
  27. # Only for Python 3.3+
  28. import lzma # noqa
  29. SUPPORTED_EXTENSIONS += XZ_EXTENSIONS
  30. except ImportError:
  31. logger.debug("lzma module is not available")
  32. def current_umask():
  33. # type: () -> int
  34. """Get the current umask which involves having to set it temporarily."""
  35. mask = os.umask(0)
  36. os.umask(mask)
  37. return mask
  38. def split_leading_dir(path):
  39. # type: (str) -> List[str]
  40. path = path.lstrip("/").lstrip("\\")
  41. if "/" in path and (
  42. ("\\" in path and path.find("/") < path.find("\\")) or "\\" not in path
  43. ):
  44. return path.split("/", 1)
  45. elif "\\" in path:
  46. return path.split("\\", 1)
  47. else:
  48. return [path, ""]
  49. def has_leading_dir(paths):
  50. # type: (Iterable[str]) -> bool
  51. """Returns true if all the paths have the same leading path name
  52. (i.e., everything is in one subdirectory in an archive)"""
  53. common_prefix = None
  54. for path in paths:
  55. prefix, rest = split_leading_dir(path)
  56. if not prefix:
  57. return False
  58. elif common_prefix is None:
  59. common_prefix = prefix
  60. elif prefix != common_prefix:
  61. return False
  62. return True
  63. def is_within_directory(directory, target):
  64. # type: (str, str) -> bool
  65. """
  66. Return true if the absolute path of target is within the directory
  67. """
  68. abs_directory = os.path.abspath(directory)
  69. abs_target = os.path.abspath(target)
  70. prefix = os.path.commonprefix([abs_directory, abs_target])
  71. return prefix == abs_directory
  72. def set_extracted_file_to_default_mode_plus_executable(path):
  73. # type: (str) -> None
  74. """
  75. Make file present at path have execute for user/group/world
  76. (chmod +x) is no-op on windows per python docs
  77. """
  78. os.chmod(path, (0o777 & ~current_umask() | 0o111))
  79. def zip_item_is_executable(info):
  80. # type: (ZipInfo) -> bool
  81. mode = info.external_attr >> 16
  82. # if mode and regular file and any execute permissions for
  83. # user/group/world?
  84. return bool(mode and stat.S_ISREG(mode) and mode & 0o111)
  85. def unzip_file(filename, location, flatten=True):
  86. # type: (str, str, bool) -> None
  87. """
  88. Unzip the file (with path `filename`) to the destination `location`. All
  89. files are written based on system defaults and umask (i.e. permissions are
  90. not preserved), except that regular file members with any execute
  91. permissions (user, group, or world) have "chmod +x" applied after being
  92. written. Note that for windows, any execute changes using os.chmod are
  93. no-ops per the python docs.
  94. """
  95. ensure_dir(location)
  96. zipfp = open(filename, "rb")
  97. try:
  98. zip = zipfile.ZipFile(zipfp, allowZip64=True)
  99. leading = has_leading_dir(zip.namelist()) and flatten
  100. for info in zip.infolist():
  101. name = info.filename
  102. fn = name
  103. if leading:
  104. fn = split_leading_dir(name)[1]
  105. fn = os.path.join(location, fn)
  106. dir = os.path.dirname(fn)
  107. if not is_within_directory(location, fn):
  108. message = (
  109. "The zip file ({}) has a file ({}) trying to install "
  110. "outside target directory ({})"
  111. )
  112. raise InstallationError(message.format(filename, fn, location))
  113. if fn.endswith("/") or fn.endswith("\\"):
  114. # A directory
  115. ensure_dir(fn)
  116. else:
  117. ensure_dir(dir)
  118. # Don't use read() to avoid allocating an arbitrarily large
  119. # chunk of memory for the file's content
  120. fp = zip.open(name)
  121. try:
  122. with open(fn, "wb") as destfp:
  123. shutil.copyfileobj(fp, destfp)
  124. finally:
  125. fp.close()
  126. if zip_item_is_executable(info):
  127. set_extracted_file_to_default_mode_plus_executable(fn)
  128. finally:
  129. zipfp.close()
  130. def untar_file(filename, location):
  131. # type: (str, str) -> None
  132. """
  133. Untar the file (with path `filename`) to the destination `location`.
  134. All files are written based on system defaults and umask (i.e. permissions
  135. are not preserved), except that regular file members with any execute
  136. permissions (user, group, or world) have "chmod +x" applied after being
  137. written. Note that for windows, any execute changes using os.chmod are
  138. no-ops per the python docs.
  139. """
  140. ensure_dir(location)
  141. if filename.lower().endswith(".gz") or filename.lower().endswith(".tgz"):
  142. mode = "r:gz"
  143. elif filename.lower().endswith(BZ2_EXTENSIONS):
  144. mode = "r:bz2"
  145. elif filename.lower().endswith(XZ_EXTENSIONS):
  146. mode = "r:xz"
  147. elif filename.lower().endswith(".tar"):
  148. mode = "r"
  149. else:
  150. logger.warning(
  151. "Cannot determine compression type for file %s",
  152. filename,
  153. )
  154. mode = "r:*"
  155. tar = tarfile.open(filename, mode, encoding="utf-8")
  156. try:
  157. leading = has_leading_dir([member.name for member in tar.getmembers()])
  158. for member in tar.getmembers():
  159. fn = member.name
  160. if leading:
  161. fn = split_leading_dir(fn)[1]
  162. path = os.path.join(location, fn)
  163. if not is_within_directory(location, path):
  164. message = (
  165. "The tar file ({}) has a file ({}) trying to install "
  166. "outside target directory ({})"
  167. )
  168. raise InstallationError(message.format(filename, path, location))
  169. if member.isdir():
  170. ensure_dir(path)
  171. elif member.issym():
  172. try:
  173. # https://github.com/python/typeshed/issues/2673
  174. tar._extract_member(member, path) # type: ignore
  175. except Exception as exc:
  176. # Some corrupt tar files seem to produce this
  177. # (specifically bad symlinks)
  178. logger.warning(
  179. "In the tar file %s the member %s is invalid: %s",
  180. filename,
  181. member.name,
  182. exc,
  183. )
  184. continue
  185. else:
  186. try:
  187. fp = tar.extractfile(member)
  188. except (KeyError, AttributeError) as exc:
  189. # Some corrupt tar files seem to produce this
  190. # (specifically bad symlinks)
  191. logger.warning(
  192. "In the tar file %s the member %s is invalid: %s",
  193. filename,
  194. member.name,
  195. exc,
  196. )
  197. continue
  198. ensure_dir(os.path.dirname(path))
  199. assert fp is not None
  200. with open(path, "wb") as destfp:
  201. shutil.copyfileobj(fp, destfp)
  202. fp.close()
  203. # Update the timestamp (useful for cython compiled files)
  204. tar.utime(member, path)
  205. # member have any execute permissions for user/group/world?
  206. if member.mode & 0o111:
  207. set_extracted_file_to_default_mode_plus_executable(path)
  208. finally:
  209. tar.close()
  210. def unpack_file(
  211. filename, # type: str
  212. location, # type: str
  213. content_type=None, # type: Optional[str]
  214. ):
  215. # type: (...) -> None
  216. filename = os.path.realpath(filename)
  217. if (
  218. content_type == "application/zip"
  219. or filename.lower().endswith(ZIP_EXTENSIONS)
  220. or zipfile.is_zipfile(filename)
  221. ):
  222. unzip_file(filename, location, flatten=not filename.endswith(".whl"))
  223. elif (
  224. content_type == "application/x-gzip"
  225. or tarfile.is_tarfile(filename)
  226. or filename.lower().endswith(TAR_EXTENSIONS + BZ2_EXTENSIONS + XZ_EXTENSIONS)
  227. ):
  228. untar_file(filename, location)
  229. else:
  230. # FIXME: handle?
  231. # FIXME: magic signatures?
  232. logger.critical(
  233. "Cannot unpack file %s (downloaded from %s, content-type: %s); "
  234. "cannot detect archive format",
  235. filename,
  236. location,
  237. content_type,
  238. )
  239. raise InstallationError(f"Cannot determine archive format of {location}")