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 #96.

Browse source
This commit is contained in:
dachshund 2013-08-28 16:40:13 -04:00
commit 13123f39aa
2 changed files with 71 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

95
tuf/client/updater.py
View file

@ -105,6 +105,7 @@
import os
import shutil
import time
import traceback
import tuf
import tuf.conf
@ -668,6 +669,9 @@ def _update_metadata(self, metadata_role, fileinfo, compression=None,
file_length=fileinfo['length']
file_hashes=fileinfo['hashes']
# A dictionary to keep the error from every mirror that we try.
mirror_errors = {}
# Attempt a file download from each mirror until the file is downloaded and
# verified. If the signature of the downloaded file is valid, proceed,
# otherwise log a warning and try the next mirror. 'metadata_file_object'
@ -677,6 +681,7 @@ def _update_metadata(self, metadata_role, fileinfo, compression=None,
# 'tuf.formats.SIGNABLE_SCHEMA'.
metadata_file_object = None
metadata_signable = None
for mirror_url in get_mirrors('meta',
metadata_filename.encode("utf-8"),
self.mirrors):
@ -684,41 +689,50 @@ def _update_metadata(self, metadata_role, fileinfo, compression=None,
metadata_file_object = \
download_file(mirror_url, file_length, file_hashes,
STRICT_REQUIRED_LENGTH=STRICT_REQUIRED_LENGTH)
except tuf.DownloadError, e:
logger.warn('Download failed from '+mirror_url+'.')
continue
if compression:
metadata_file_object.decompress_temp_file_object(compression)
# Read and load the downloaded file.
metadata_signable = tuf.util.load_json_string(metadata_file_object.read())
# Verify the signature on the downloaded metadata object.
try:
valid = tuf.sig.verify(metadata_signable, metadata_role)
except (tuf.UnknownRoleError, tuf.FormatError, tuf.Error), e:
# FIXME: Exception.message is deprecated in 2.6, and gone in 3.0,
# but this is a workaround for Unicode messages. We need a long-term
# solution with #61.
# http://bugs.python.org/issue2517
message = 'Unable to verify '+metadata_filename+':'+\
e.message.encode("utf-8")
logger.exception(message)
metadata_signable = None
continue
except:
logger.exception('Download failed from '+mirror_url+'.')
mirror_errors[mirror_url] = traceback.format_exc(1)
else:
if valid:
logger.debug('Good signature on '+mirror_url+'.')
break
else:
logger.warn('Bad signature on '+mirror_url+'.')
if compression:
metadata_file_object.decompress_temp_file_object(compression)
# Read and load the downloaded file.
try:
metadata_signable = \
tuf.util.load_json_string(metadata_file_object.read())
except:
logger.exception('Invalid metadata from '+mirror_url+'.')
mirror_errors[mirror_url] = traceback.format_exc(1)
metadata_signable = None
continue
else:
# Verify the signature on the downloaded metadata object.
try:
valid = tuf.sig.verify(metadata_signable, metadata_role)
except (tuf.UnknownRoleError, tuf.FormatError, tuf.Error), e:
# FIXME: Exception.message is deprecated in 2.6, and gone in 3.0,
# but this is a workaround for Unicode messages. We need a
# long-term solution with #61.
# http://bugs.python.org/issue2517
message = 'Unable to verify '+metadata_filename+':'+\
e.message.encode("utf-8")
logger.exception(message)
mirror_errors[mirror_url] = message
metadata_signable = None
else:
if valid:
logger.debug('Good signature on '+mirror_url+'.')
break
else:
message = 'Bad signature on '+mirror_url+'.'
logger.warn(message)
mirror_errors[mirror_url] = message
metadata_signable = None
# Raise an exception if a valid metadata signable could not be downloaded
# from any of the mirrors.
if metadata_signable is None:
message = 'Unable to update '+repr(metadata_filename)+'.'
message = 'Unable to update '+str(metadata_filename)+\
' from all known mirrors: '+str(mirror_errors)
logger.error(message)
raise tuf.RepositoryError(message)
@ -1867,7 +1881,7 @@ def download_target(self, target, destination_directory):
# Raise 'tuf.FormatError' if the check fail.
tuf.formats.TARGETFILE_SCHEMA.check_match(target)
tuf.formats.PATH_SCHEMA.check_match(destination_directory)
# Reference to the 'get_list_of_mirrors' function.
get_mirrors = tuf.mirrors.get_list_of_mirrors
@ -1879,7 +1893,14 @@ def download_target(self, target, destination_directory):
trusted_length = target['fileinfo']['length']
trusted_hashes = target['fileinfo']['hashes']
# A dictionary to keep the error from every mirror that we try.
mirror_errors = {}
# Wherein the downloaded target file will be stored.
target_file_object = None
logger.info('Trying to download: '+str(target_filepath))
# Iterate through the repositority mirrors until we successfully
# download a target.
for mirror_url in get_mirrors('target', target_filepath, self.mirrors):
@ -1887,13 +1908,19 @@ def download_target(self, target, destination_directory):
target_file_object = download_file(mirror_url, trusted_length,
trusted_hashes)
break
except (tuf.DownloadError, tuf.FormatError), e:
logger.warn('Download failed from '+mirror_url+'.')
except:
# Remember the error from this mirror, and "reset" the target file.
logger.exception('Download failed from '+mirror_url+'.')
mirror_errors[mirror_url] = traceback.format_exc(1)
target_file_object = None
continue
# We have gone through all the mirrors. Did we get a target file object?
if target_file_object == None:
raise tuf.DownloadError('No download locations known.')
raise tuf.DownloadError('Failed to download target '+\
str(target_filepath)+\
' from all known mirrors: '+\
str(mirror_errors))
# We acquired a target file object from a mirror. Move the file into
# place (i.e., locally to 'destination_directory').

17
tuf/tests/system_tests/test_endless_data_attack.py Normal file → Executable file
View file

@ -31,8 +31,6 @@
"""
# TODO:...
import os
import shutil
import urllib
@ -41,7 +39,7 @@
import tuf
from tuf.interposition import urllib_tuf
from tuf.log import logger
class EndlessDataAttack(Exception):
pass
@ -91,7 +89,6 @@ def test_arbitrary_package_attack(TUF=False, TIMESTAMP=False):
if TUF:
# Update TUF metadata before attacker modifies anything.
util_test_tools.tuf_refresh_repo(root_repo, keyids)
# Modify the url. Remember that the interposition will intercept
# urls that have 'localhost:9999' hostname, which was specified in
# the json interposition configuration file. Look for 'hostname'
@ -102,10 +99,12 @@ def test_arbitrary_package_attack(TUF=False, TIMESTAMP=False):
# Attacker modifies the file at the targets repository.
target = os.path.join(tuf_targets, file_basename)
util_test_tools.modify_file_at_repository(target, endless_data)
# Attacker modifies the timestamp.txt metadata.
if TIMESTAMP:
metadata = os.path.join(tuf_repo, 'metadata')
timestamp = os.path.join(metadata, 'timestamp.txt')
# FIXME: This does not correctly "patch" the timestamp metadata.
util_test_tools.modify_file_at_repository(timestamp, endless_data)
# Attacker modifies the file at the regular repository.
@ -118,11 +117,11 @@ def test_arbitrary_package_attack(TUF=False, TIMESTAMP=False):
# Client downloads (tries to download) the file.
_download(url=url_to_repo, filename=downloaded_file, tuf=TUF)
except (tuf.DownloadError,tuf.RepositoryError), e:
except (tuf.DownloadError, tuf.RepositoryError), e:
# If tuf.DownloadError or tuf.RepositoryError is raised, this means
# that TUF has prevented the download of an unrecognized file. Enable
# the logging to see, what actually happened.
pass
# logging to see what actually happened.
logger.warn('Download failed: '+repr(e))
else:
# Check whether the attack succeeded by inspecting the content of the
@ -157,7 +156,11 @@ def test_arbitrary_package_attack(TUF=False, TIMESTAMP=False):
try:
# FIXME: This test passes, but not yet because we avoided an endless data
# attack with timestamp metadata, but rather because the timestamp metadata
# is invalid.
test_arbitrary_package_attack(TUF=True, TIMESTAMP=True)
raise EndlessDataAttack('Timestamp metadata is not yet immune from the endless data attack!')
except EndlessDataAttack, error:
print('With TUF: '+str(error))