Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_bad_wheel_filename(filename):
exc = pytest.raises(WheelError, WheelFile, filename)
exc.match('^Bad wheel filename {!r}$'.format(filename))
def pack(directory, dest_dir):
"""Repack a previously unpacked wheel directory into a new wheel file.
The .dist-info/WHEEL file must contain one or more tags so that the target
wheel file name can be determined.
:param directory: The unpacked wheel directory
:param dest_dir: Destination directory (defaults to the current directory)
"""
# Find the .dist-info directory
dist_info_dirs = [fn for fn in os.listdir(directory)
if os.path.isdir(os.path.join(directory, fn)) and DIST_INFO_RE.match(fn)]
if len(dist_info_dirs) > 1:
raise WheelError('Multiple .dist-info directories found in {}'.format(directory))
elif not dist_info_dirs:
raise WheelError('No .dist-info directories found in {}'.format(directory))
# Determine the target wheel filename
dist_info_dir = dist_info_dirs[0]
name_version = DIST_INFO_RE.match(dist_info_dir).group('namever')
# Read the tags from .dist-info/WHEEL
with open(os.path.join(directory, dist_info_dir, 'WHEEL')) as f:
tags = [line.split(' ')[1].rstrip() for line in f if line.startswith('Tag: ')]
if not tags:
raise WheelError('No tags present in {}/WHEEL; cannot determine target wheel filename'
.format(dist_info_dir))
# Reassemble the tags for the wheel file
impls = sorted({tag.split('-')[0] for tag in tags})
elif not dist_info_dirs:
raise WheelError('No .dist-info directories found in {}'.format(directory))
# Determine the target wheel filename
dist_info_dir = dist_info_dirs[0]
name_version = DIST_INFO_RE.match(dist_info_dir).group('namever')
# Add the build number if specific
if build_number:
name_version += '-' + build_number
# Read the tags from .dist-info/WHEEL
with open(os.path.join(directory, dist_info_dir, 'WHEEL')) as f:
tags = [line.split(' ')[1].rstrip() for line in f if line.startswith('Tag: ')]
if not tags:
raise WheelError('No tags present in {}/WHEEL; cannot determine target wheel filename'
.format(dist_info_dir))
# Reassemble the tags for the wheel file
impls = sorted({tag.split('-')[0] for tag in tags})
abivers = sorted({tag.split('-')[1] for tag in tags})
platforms = sorted({tag.split('-')[2] for tag in tags})
tagline = '-'.join(['.'.join(impls), '.'.join(abivers), '.'.join(platforms)])
# Repack the wheel
wheel_path = os.path.join(dest_dir, '{}-{}.whl'.format(name_version, tagline))
with WheelFile(wheel_path, 'w') as wf:
print("Repacking wheel as {}...".format(wheel_path), end='')
sys.stdout.flush()
wf.write_files(directory)
print('OK')
def _update_crc(newdata, eof=None):
if eof is None:
eof = ef._eof
update_crc_orig(newdata)
else: # Python 2
update_crc_orig(newdata, eof)
running_hash.update(newdata)
if eof and running_hash.digest() != expected_hash:
raise WheelError("Hash mismatch for file '{}'".format(native(ef_name)))
def open(self, name_or_info, mode="r", pwd=None):
def _update_crc(newdata, eof=None):
if eof is None:
eof = ef._eof
update_crc_orig(newdata)
else: # Python 2
update_crc_orig(newdata, eof)
running_hash.update(newdata)
if eof and running_hash.digest() != expected_hash:
raise WheelError("Hash mismatch for file '{}'".format(native(ef_name)))
ef_name = as_unicode(name_or_info.filename if isinstance(name_or_info, ZipInfo)
else name_or_info)
if mode == 'r' and not ef_name.endswith('/') and ef_name not in self._file_hashes:
raise WheelError("No hash found for file '{}'".format(native(ef_name)))
ef = ZipFile.open(self, name_or_info, mode, pwd)
if mode == 'r' and not ef_name.endswith('/'):
algorithm, expected_hash = self._file_hashes[ef_name]
if expected_hash is not None:
# Monkey patch the _update_crc method to also check for the hash from RECORD
running_hash = hashlib.new(algorithm)
update_crc_orig, ef._update_crc = ef._update_crc, _update_crc
return ef
def __init__(self, file, mode='r'):
basename = os.path.basename(file)
self.parsed_filename = WHEEL_INFO_RE.match(basename)
if not basename.endswith('.whl') or self.parsed_filename is None:
raise WheelError("Bad wheel filename {!r}".format(basename))
ZipFile.__init__(self, file, mode, compression=ZIP_DEFLATED, allowZip64=True)
self.dist_info_path = '{}.dist-info'.format(self.parsed_filename.group('namever'))
self.record_path = self.dist_info_path + '/RECORD'
self._file_hashes = OrderedDict()
self._file_sizes = {}
if mode == 'r':
# Ignore RECORD and any embedded wheel signatures
self._file_hashes[self.record_path] = None, None
self._file_hashes[self.record_path + '.jws'] = None, None
self._file_hashes[self.record_path + '.p7s'] = None, None
# Fill in the expected hashes by reading them from RECORD
try:
record = self.open(self.record_path)
def _update_crc(newdata, eof=None):
if eof is None:
eof = ef._eof
update_crc_orig(newdata)
else: # Python 2
update_crc_orig(newdata, eof)
running_hash.update(newdata)
if eof and running_hash.digest() != expected_hash:
raise WheelError("Hash mismatch for file '{}'".format(native(ef_name)))
def _update_crc(newdata, eof=None):
if eof is None:
eof = ef._eof
update_crc_orig(newdata)
else: # Python 2
update_crc_orig(newdata, eof)
running_hash.update(newdata)
if eof and running_hash.digest() != expected_hash:
raise WheelError("Hash mismatch for file '{}'".format(native(ef_name)))
if eof is None:
eof = ef._eof
update_crc_orig(newdata)
else: # Python 2
update_crc_orig(newdata, eof)
running_hash.update(newdata)
if eof and running_hash.digest() != expected_hash:
raise WheelError("Hash mismatch for file '{}'".format(native(ef_name)))
ef = ZipFile.open(self, name_or_info, mode, pwd)
ef_name = as_unicode(name_or_info.filename if isinstance(name_or_info, ZipInfo)
else name_or_info)
if mode == 'r' and not ef_name.endswith('/'):
if ef_name not in self._file_hashes:
raise WheelError("No hash found for file '{}'".format(native(ef_name)))
algorithm, expected_hash = self._file_hashes[ef_name]
if expected_hash is not None:
# Monkey patch the _update_crc method to also check for the hash from RECORD
running_hash = hashlib.new(algorithm)
update_crc_orig, ef._update_crc = ef._update_crc, _update_crc
return ef