Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
shutil.move(self.client_metadata_previous, previous_backup)
self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository1',
self.repository_mirrors)
shutil.move(previous_backup, self.client_metadata_previous)
# Test: repository missing the required 'root.json' file.
client_root_file = os.path.join(self.client_metadata_current, 'root.json')
backup_root_file = client_root_file + '.backup'
shutil.move(client_root_file, backup_root_file)
self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository1',
self.repository_mirrors)
# Restore the client's 'root.json file.
shutil.move(backup_root_file, client_root_file)
# Test: Normal 'tuf.client.updater.Updater' instantiation.
updater.Updater('test_repository1', self.repository_mirrors)
# 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'.
repository_basepath = self.repository_directory[len(os.getcwd()):]
url_prefix = \
'http://localhost:' + str(self.SERVER_PORT) + repository_basepath
# Setting 'tuf.settings.repository_directory' with the temporary client
# directory copied from the original repository files.
tuf.settings.repositories_directory = self.client_directory
self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix,
'metadata_path': 'metadata',
'targets_path': 'targets',
'confined_target_dirs': ['']}}
# Create the repository instance. The test cases will use this client
# updater to refresh metadata, fetch target files, etc.
self.repository_updater = updater.Updater(self.repository_name,
self.repository_mirrors)
repository_basepath = self.repository_directory[len(os.getcwd()):]
url_prefix = \
'http://localhost:' + str(self.SERVER_PORT) + repository_basepath
# Setting 'tuf.settings.repository_directory' with the temporary client
# directory copied from the original repository files.
tuf.settings.repositories_directory = self.client_directory
self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix,
'metadata_path': 'metadata',
'targets_path': 'targets',
'confined_target_dirs': ['']}}
# Creating a repository instance. The test cases will use this client
# updater to refresh metadata, fetch target files, etc.
self.repository_updater = updater.Updater(self.repository_name,
self.repository_mirrors)
# Metadata role keys are needed by the test cases to make changes to the
# repository (e.g., adding a new target file to 'targets.json' and then
# requesting a refresh()).
self.role_keys = _load_role_keys(self.keystore_directory)
def do_update(self):
# Client side repository.
tuf_client = os.path.join(self.root_repo, 'tuf_client')
downloads_dir = os.path.join(self.root_repo, 'downloads')
# Adjust client's configuration file.
tuf.conf.repository_directory = tuf_client
updater = tuf.client.updater.Updater('my_repo', self.mirrors)
# Refresh the repository's top-level roles, store the target information for
# all the targets tracked, and determine which of these targets have been
# updated.
updater.refresh()
# Obtain a list of available targets.
targets = []
relative_target_filepaths = self.relpath_from_targets(self.target_filepaths)
for target_filepath in relative_target_filepaths:
target_info = updater.target(target_filepath)
targets.append(target_info)
# Download each of these updated targets and save them locally.
updated_targets = updater.updated_targets(targets, downloads_dir)
for target in updated_targets:
message = 'The repository mirror supplied is invalid.'
raise tuf.RepositoryError(message)
# Set the local repository directory containing all of the metadata files.
tuf.conf.repository_directory = '.'
# Set the repository mirrors. This dictionary is needed by the Updater
# class of updater.py.
repository_mirrors = {'mirror': {'url_prefix': repository_mirror,
'metadata_path': 'repository',
'targets_path': 'repository/targets',
'confined_target_dirs': ['']}}
# Create the repository object using the repository name 'repository'
# and the repository mirrors defined above.
updater = tuf.client.updater.Updater('repository', repository_mirrors)
# The local destination directory to save the target files.
destination_directory = './targets'
# Refresh the repository's top-level roles, store the target information for
# all the targets tracked, and determine which of these targets have been
# updated.
updater.refresh()
all_targets = updater.all_targets()
updated_targets = updater.updated_targets(all_targets, destination_directory)
# Download each of these updated targets and save them locally.
for target in updated_targets:
try:
updater.download_target(target, destination_directory)
except tuf.DownloadError as e:
# 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'.
repository_basepath = self.repository_directory[len(os.getcwd()):]
url_prefix = \
'http://localhost:' + str(self.SERVER_PORT) + repository_basepath
# Setting 'tuf.settings.repository_directory' with the temporary client
# directory copied from the original repository files.
tuf.settings.repositories_directory = self.client_directory
self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix,
'metadata_path': 'metadata',
'targets_path': 'targets',
'confined_target_dirs': ['']}}
# Create the repository instance. The test cases will use this client
# updater to refresh metadata, fetch target files, etc.
self.repository_updater = updater.Updater(self.repository_name,
self.repository_mirrors)
# While 0.3s has consistently worked on Travis and local builds, it led to
# occasional failures in AppVeyor builds, so increasing this to 2s, sadly.
time.sleep(2)
# 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'.
repository_basepath = self.repository_directory[len(os.getcwd()):]
url_prefix = \
'http://localhost:' + str(SERVER_PORT) + repository_basepath
self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix,
'metadata_path': 'metadata', 'targets_path': 'targets',
'confined_target_dirs': ['']}}
# Creating a repository instance. The test cases will use this client
# updater to refresh metadata, fetch target files, etc.
self.repository_updater = updater.Updater(self.repository_name,
self.repository_mirrors)
# Extract the file information of the targets specified in 'targets.json'.
self.repository_updater.refresh()
targets_metadata = self.repository_updater.metadata['current']['targets']
target_files = targets_metadata['targets']
# Extract random target from 'target_files', which will be compared to what
# is returned by get_one_valid_targetinfo(). Restore the popped target
# (dict value stored in the metadata store) so that it can be found later.
filepath, fileinfo = target_files.popitem()
target_files[filepath] = fileinfo
target_targetinfo = self.repository_updater.get_one_valid_targetinfo(filepath)
self.assertTrue(tuf.formats.TARGETINFO_SCHEMA.matches(target_targetinfo))
self.assertEqual(target_targetinfo['filepath'], filepath)
#tuf.log.set_log_level(logging.DEBUG)
# Set the local repository directory containing the metadata files.
tuf.conf.repository_directory = '.'
# Set the repository mirrors. This dictionary is needed by the Updater
# class of updater.py. The client will download metadata and target
# files from any one of these mirrors.
repository_mirrors = {'mirror1': {'url_prefix': 'http://localhost:8001',
'metadata_path': 'metadata',
'targets_path': 'targets',
'confined_target_paths': ['']}}
# Create the Upater object using the updater name 'tuf-example'
# and the repository mirrors defined above.
updater = tuf.client.updater.Updater('tuf-example', repository_mirrors)
# Set the local destination directory to save the target files.
destination_directory = './targets'
# Refresh the repository's top-level roles, store the target information for
# all the targets tracked, and determine which of these targets have been
# updated.
updater.refresh()
all_targets = updater.all_targets()
updated_targets = updater.updated_targets(all_targets, destination_directory)
# Download each of these updated targets and save them locally.
for target in updated_targets:
try:
updater.download_target(target, destination_directory)
except tuf.DownloadError, e:
def run(self):
"""
Check for updates
"""
if not self.mirrors:
return
while True:
try:
tuf.conf.repository_directory = os.path.join(self.bundle_path,
'repo')
updater = tuf.client.updater.Updater('leap-updater',
self.mirrors)
updater.refresh()
targets = updater.all_targets()
updated_targets = updater.updated_targets(targets,
self.source_path)
if updated_targets:
logger.info("There is updates needed. Start downloading "
"updates.")
for target in updated_targets:
updater.download_target(target, self.dest_path)
self._set_permissions(target)
if os.path.isdir(self.dest_path):
if os.path.isdir(self.update_path):
shutil.rmtree(self.update_path)
shutil.move(self.dest_path, self.update_path)
try:
tuf.formats.URL_SCHEMA.check_match(uptane_tuf_server.url)
except tuf.FormatError as e:
message = 'The repository mirror supplied is invalid.'
raise tuf.RepositoryError(message)
# Set the repository mirrors. This dictionary is needed by the Updater
# class of updater.py.
repository_mirrors = {'mirror': {'url_prefix': uptane_tuf_server.url,
'metadata_path': 'metadata',
'targets_path': 'targets',
'confined_target_dirs': ['']}}
# Create the repository object using the repository name 'repository'
# and the repository mirrors defined above.
updater = tuf.client.updater.Updater('repository', repository_mirrors)
# Refresh top-level metadata.
updater.refresh()
# Fetch the target info specified by the director role.
# (Note that this info is not final; it's simply what the director
# specifies. When we call updater.target() on each file path, that is when
# all delegation is validated.)
directed_target_info = updater.targets_of_role('director')
directed_filepaths = [f['filepath'] for f in directed_target_info]
trustworthy_directed_target_info = []
trustworthy_directed_filepaths = []
failed_targets = False