Elgato_dark/python-tuf
1
0
Fork 0
mirror of https://github.com/theupdateframework/python-tuf synced 2026-05-24 10:08:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1202 from joshuagl/joshuagl/updater-simplify

Browse source 
Simplify updater logic for downloading and verifying target files
This commit is contained in:
lukpueh 2020-11-26 13:48:35 +01:00 committed by GitHub
commit e061bc6ebb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 33 additions and 192 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
tests/test_updater.py
View file

@ -1617,38 +1617,19 @@ def test_9__get_target_hash(self):
def test_10__hard_check_file_length(self):
def test_10__check_file_length(self):
# Test for exception if file object is not equal to trusted file length.
with tempfile.TemporaryFile() as temp_file_object:
temp_file_object.write(b'X')
temp_file_object.seek(0)
self.assertRaises(tuf.exceptions.DownloadLengthMismatchError,
self.repository_updater._hard_check_file_length,
self.repository_updater._check_file_length,
temp_file_object, 10)
def test_10__soft_check_file_length(self):
# Test for exception if file object is not equal to trusted file length.
with tempfile.TemporaryFile() as temp_file_object:
temp_file_object.write(b'XXX')
temp_file_object.seek(0)
self.assertRaises(tuf.exceptions.DownloadLengthMismatchError,
self.repository_updater._soft_check_file_length,
temp_file_object, 1)
# Verify that an exception is not raised if the file length <= the observed
# file length.
temp_file_object.seek(0)
self.repository_updater._soft_check_file_length(temp_file_object, 3)
temp_file_object.seek(0)
self.repository_updater._soft_check_file_length(temp_file_object, 4)
def test_10__targets_of_role(self):
# Test for non-existent role.
self.assertRaises(tuf.exceptions.UnknownRoleError,
@ -1764,26 +1745,6 @@ def test_11__verify_metadata_file(self):
metadata_file_object, 'root')
def test_12__get_file(self):
# Test for an "unsafe" download, where the file is downloaded up to
# a required length (and no more). The "safe" download approach
# downloads an exact required length.
targets_path = os.path.join(self.repository_directory, 'metadata', 'targets.json')
file_size, file_hashes = securesystemslib.util.get_file_details(targets_path)
file_type = 'meta'
def verify_target_file(targets_path):
# Every target file must have its length and hashes inspected.
self.repository_updater._hard_check_file_length(targets_path, file_size)
self.repository_updater._check_hashes(targets_path, file_hashes)
self.repository_updater._get_file('targets.json', verify_target_file,
file_type, file_size, download_safely=True).close()
self.repository_updater._get_file('targets.json', verify_target_file,
file_type, file_size, download_safely=False).close()
def test_13__targets_of_role(self):
# Test case where a list of targets is given. By default, the 'targets'
# parameter is None.

182
tuf/client/updater.py
View file

@ -1206,7 +1206,7 @@ def _check_hashes(self, file_object, trusted_hashes):
def _hard_check_file_length(self, file_object, trusted_file_length):
def _check_file_length(self, file_object, trusted_file_length):
"""
<Purpose>
Non-public method that ensures the length of 'file_object' is strictly
@ -1233,8 +1233,10 @@ def _hard_check_file_length(self, file_object, trusted_file_length):
None.
"""
file_object.seek(0)
observed_length = len(file_object.read())
# seek to the end of the file; that is offset 0 from the end of the file,
# represented by a whence value of 2
file_object.seek(0, 2)
observed_length = file_object.tell()
# Return and log a message if the length 'file_object' is equal to
# 'trusted_file_length', otherwise raise an exception. A hard check
@ -1252,53 +1254,6 @@ def _hard_check_file_length(self, file_object, trusted_file_length):
def _soft_check_file_length(self, file_object, trusted_file_length):
"""
<Purpose>
Non-public method that checks the trusted file length of a file object.
The length of the file must be less than or equal to the expected
length. This is a deliberately redundant implementation designed to
complement tuf.download._check_downloaded_length().
<Arguments>
file_object:
A file object.
trusted_file_length:
A non-negative integer that is the trusted length of the file.
<Exceptions>
tuf.exceptions.DownloadLengthMismatchError, if the lengths do
not match.
<Side Effects>
Reads the contents of 'file_object' and logs a message if 'file_object'
is less than or equal to the trusted length.
Position within file_object is changed.
<Returns>
None.
"""
# Read the entire contents of 'file_object', a
file_object.seek(0)
observed_length = len(file_object.read())
# Return and log a message if 'file_object' is less than or equal to
# 'trusted_file_length', otherwise raise an exception. A soft check
# ensures that an upper bound restricts how large a file is downloaded.
if observed_length > trusted_file_length:
raise tuf.exceptions.DownloadLengthMismatchError(trusted_file_length,
observed_length)
else:
logger.debug('Observed length (' + str(observed_length) +\
') <= trusted length (' + str(trusted_file_length) + ')')
def _get_target_file(self, target_filepath, file_length, file_hashes,
prefix_filename_with_hash):
"""
@ -1340,15 +1295,6 @@ def _get_target_file(self, target_filepath, file_length, file_hashes,
A file object containing the target.
"""
# Define a callable function that is passed as an argument to _get_file()
# and called. The 'verify_target_file' function ensures the file length
# and hashes of 'target_filepath' are strictly equal to the trusted values.
def verify_target_file(target_file_object):
# Every target file must have its length and hashes inspected.
self._hard_check_file_length(target_file_object, file_length)
self._check_hashes(target_file_object, file_hashes)
if self.consistent_snapshot and prefix_filename_with_hash:
# Note: values() does not return a list in Python 3. Use list()
# on values() for Python 2+3 compatibility.
@ -1356,8 +1302,32 @@ def verify_target_file(target_file_object):
dirname, basename = os.path.split(target_filepath)
target_filepath = os.path.join(dirname, target_digest + '.' + basename)
return self._get_file(target_filepath, verify_target_file,
'target', file_length, download_safely=True)
file_mirrors = tuf.mirrors.get_list_of_mirrors('target', target_filepath,
self.mirrors)
# file_mirror (URL): error (Exception)
file_mirror_errors = {}
file_object = None
for file_mirror in file_mirrors:
try:
file_object = tuf.download.safe_download(file_mirror, file_length)
# Verify 'file_object' against the expected length and hashes.
self._check_file_length(file_object, file_length)
self._check_hashes(file_object, file_hashes)
# If the file verifies, we don't need to try more mirrors
return file_object
except Exception as exception:
# Remember the error from this mirror, and "reset" the target file.
logger.debug('Update failed from ' + file_mirror + '.')
file_mirror_errors[file_mirror] = exception
file_object = None
logger.debug('Failed to update ' + repr(target_filepath) + ' from'
' all mirrors: ' + repr(file_mirror_errors))
raise tuf.exceptions.NoWorkingMirrorError(file_mirror_errors)
@ -1641,96 +1611,6 @@ def _get_metadata_file(self, metadata_role, remote_filename,
def _get_file(self, filepath, verify_file_function, file_type, file_length,
download_safely=True):
"""
<Purpose>
Non-public method that tries downloading, up to a certain length, a
metadata or target file from a list of known mirrors. As soon as the first
valid copy of the file is found, the rest of the mirrors will be skipped.
<Arguments>
filepath:
The relative metadata or target filepath.
verify_file_function:
A callable function that expects a file object and raises an exception
if the file is invalid.
Target files and uncompressed versions of metadata may be verified with
'verify_file_function'.
file_type:
Type of data needed for download, must correspond to one of the strings
in the list ['meta', 'target']. 'meta' for metadata file type or
'target' for target file type. It should correspond to the
'securesystemslib.formats.NAME_SCHEMA' format.
file_length:
The expected length, or upper bound, of the target or metadata file to
be downloaded.
download_safely:
A boolean switch to toggle safe or unsafe download of the file.
<Exceptions>
tuf.exceptions.NoWorkingMirrorError:
The metadata could not be fetched. This is raised only when all known
mirrors failed to provide a valid copy of the desired metadata file.
<Side Effects>
The file is downloaded from all known repository mirrors in the worst
case. If a valid copy of the file is found, it is stored in a temporary
file and returned.
<Returns>
A file object containing the metadata or target.
"""
file_mirrors = tuf.mirrors.get_list_of_mirrors(file_type, filepath,
self.mirrors)
# file_mirror (URL): error (Exception)
file_mirror_errors = {}
file_object = None
for file_mirror in file_mirrors:
try:
# TODO: Instead of the more fragile 'download_safely' switch, unroll
# the function into two separate ones: one for "safe" download, and the
# other one for "unsafe" download? This should induce safer and more
# readable code.
if download_safely:
file_object = tuf.download.safe_download(file_mirror, file_length)
else:
file_object = tuf.download.unsafe_download(file_mirror, file_length)
# Verify 'file_object' according to the callable function.
# 'file_object' is also verified if decompressed above (i.e., the
# uncompressed version).
verify_file_function(file_object)
except Exception as exception:
# Remember the error from this mirror, and "reset" the target file.
logger.debug('Update failed from ' + file_mirror + '.')
file_mirror_errors[file_mirror] = exception
file_object = None
else:
break
if file_object:
return file_object
else:
logger.debug('Failed to update ' + repr(filepath) + ' from'
' all mirrors: ' + repr(file_mirror_errors))
raise tuf.exceptions.NoWorkingMirrorError(file_mirror_errors)
def _update_metadata(self, metadata_role, upperbound_filelength, version=None):
"""
<Purpose>