mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
Merge pull request #540 from vladimir-v-diaz/pylint
Resolve remaining Pylint warnings
This commit is contained in:
Vladimir Diaz
GitHub
commit
7ed5d2dfee
10 changed files with 149 additions and 162 deletions
|
|
@ -10,16 +10,16 @@ env:
|
|||
- TOXENV=py36
|
||||
|
||||
before_script:
|
||||
- pip install -U pip wheel tox
|
||||
- pip install -U pip wheel pylint
|
||||
- pip install -U pip wheel bandit
|
||||
- pip install -U pip wheel tox
|
||||
|
||||
script:
|
||||
- pylint tuf
|
||||
- bandit -r tuf
|
||||
- tox
|
||||
|
||||
after_success:
|
||||
- pylint tuf
|
||||
- cd tests
|
||||
- coveralls
|
||||
- cd -
|
||||
|
|
|
|||
2
pylintrc
2
pylintrc
|
|
@ -344,7 +344,7 @@ defining-attr-methods=__init__,__new__,setUp
|
|||
|
||||
# List of member names, which should be excluded from the protected access
|
||||
# warning.
|
||||
exclude-protected=_asdict, _fields, _replace, _source, _make, _generate_and_write_metadata, _delete_obsolete_metadata, _log_status_of_top_level_roles, _load_top_level_metadata, _strip_version_number, _delegated_roles
|
||||
exclude-protected=_asdict, _fields, _replace, _source, _make, _generate_and_write_metadata, _delete_obsolete_metadata, _log_status_of_top_level_roles, _load_top_level_metadata, _strip_version_number, _delegated_roles, _remove_invalid_and_duplicate_signatures
|
||||
|
||||
# List of valid names for the first argument in a class method.
|
||||
valid-classmethod-first-arg=cls
|
||||
|
|
|
|||
|
|
@ -86,22 +86,22 @@ def test_create_new_project(self):
|
|||
|
||||
self.assertTrue(isinstance(project, developer_tool.Project))
|
||||
self.assertTrue(project.layout_type == 'repo-like')
|
||||
self.assertTrue(project._prefix == location_in_repository)
|
||||
self.assertTrue(project._project_name == project_name)
|
||||
self.assertTrue(project._metadata_directory ==
|
||||
self.assertTrue(project.prefix == location_in_repository)
|
||||
self.assertTrue(project.project_name == project_name)
|
||||
self.assertTrue(project.metadata_directory ==
|
||||
os.path.join(metadata_directory,METADATA_DIRECTORY_NAME))
|
||||
self.assertTrue(project._targets_directory ==
|
||||
self.assertTrue(project.targets_directory ==
|
||||
os.path.join(metadata_directory,TARGETS_DIRECTORY_NAME))
|
||||
|
||||
# Create a blank project without a prefix.
|
||||
project = developer_tool.create_new_project(project_name, metadata_directory)
|
||||
self.assertTrue(isinstance(project, developer_tool.Project))
|
||||
self.assertTrue(project.layout_type == 'repo-like')
|
||||
self.assertTrue(project._prefix == '')
|
||||
self.assertTrue(project._project_name == project_name)
|
||||
self.assertTrue(project._metadata_directory ==
|
||||
self.assertTrue(project.prefix == '')
|
||||
self.assertTrue(project.project_name == project_name)
|
||||
self.assertTrue(project.metadata_directory ==
|
||||
os.path.join(metadata_directory,METADATA_DIRECTORY_NAME))
|
||||
self.assertTrue(project._targets_directory ==
|
||||
self.assertTrue(project.targets_directory ==
|
||||
os.path.join(metadata_directory,TARGETS_DIRECTORY_NAME))
|
||||
|
||||
# Create a blank project without a valid metadata directory.
|
||||
|
|
@ -120,10 +120,10 @@ def test_create_new_project(self):
|
|||
location_in_repository, targets_directory)
|
||||
self.assertTrue(isinstance(project, developer_tool.Project))
|
||||
self.assertTrue(project.layout_type == 'flat')
|
||||
self.assertTrue(project._prefix == location_in_repository)
|
||||
self.assertTrue(project._project_name == project_name)
|
||||
self.assertTrue(project._metadata_directory == metadata_directory)
|
||||
self.assertTrue(project._targets_directory == targets_directory)
|
||||
self.assertTrue(project.prefix == location_in_repository)
|
||||
self.assertTrue(project.project_name == project_name)
|
||||
self.assertTrue(project.metadata_directory == metadata_directory)
|
||||
self.assertTrue(project.targets_directory == targets_directory)
|
||||
|
||||
# Finally, check that if targets_directory is set, it is valid.
|
||||
self.assertRaises(securesystemslib.exceptions.FormatError, developer_tool.create_new_project,
|
||||
|
|
@ -143,10 +143,10 @@ def test_create_new_project(self):
|
|||
|
||||
self.assertTrue(isinstance(project, developer_tool.Project))
|
||||
self.assertTrue(project.layout_type == 'flat')
|
||||
self.assertTrue(project._prefix == location_in_repository)
|
||||
self.assertTrue(project._project_name == project_name)
|
||||
self.assertTrue(project._metadata_directory == metadata_directory)
|
||||
self.assertTrue(project._targets_directory == targets_directory)
|
||||
self.assertTrue(project.prefix == location_in_repository)
|
||||
self.assertTrue(project.project_name == project_name)
|
||||
self.assertTrue(project.metadata_directory == metadata_directory)
|
||||
self.assertTrue(project.targets_directory == targets_directory)
|
||||
self.assertTrue(len(project.keys) == 1)
|
||||
self.assertTrue(project.keys[0] == project_key['keyid'])
|
||||
|
||||
|
|
@ -205,7 +205,7 @@ def test_load_project(self):
|
|||
|
||||
# Load a project overwriting the prefix.
|
||||
project = developer_tool.load_project(repo_filepath, prefix='new')
|
||||
self.assertTrue(project._prefix == 'new')
|
||||
self.assertTrue(project.prefix == 'new')
|
||||
|
||||
# Load a project with a file missing.
|
||||
file_to_corrupt = os.path.join(repo_filepath, 'test-flat.json')
|
||||
|
|
@ -333,7 +333,7 @@ def test_write(self):
|
|||
|
||||
# + backup delegations.
|
||||
delegations_backup = \
|
||||
tuf.roledb.get_delegated_rolenames(project._project_name)
|
||||
tuf.roledb.get_delegated_rolenames(project.project_name)
|
||||
|
||||
# + backup layout type.
|
||||
layout_type_backup = project.layout_type
|
||||
|
|
@ -343,10 +343,10 @@ def test_write(self):
|
|||
delegation_keys_backup = project('delegation').keys
|
||||
|
||||
# + backup the prefix.
|
||||
prefix_backup = project._prefix
|
||||
prefix_backup = project.prefix
|
||||
|
||||
# + backup the name.
|
||||
name_backup = project._project_name
|
||||
name_backup = project.project_name
|
||||
|
||||
# Write and reload.
|
||||
self.assertRaises(securesystemslib.exceptions.Error, project.write)
|
||||
|
|
@ -356,17 +356,17 @@ def test_write(self):
|
|||
|
||||
# Check against backup.
|
||||
self.assertEqual(list(project.target_files.keys()), list(targets_backup.keys()))
|
||||
new_delegations = tuf.roledb.get_delegated_rolenames(project._project_name)
|
||||
new_delegations = tuf.roledb.get_delegated_rolenames(project.project_name)
|
||||
self.assertEqual(new_delegations, delegations_backup)
|
||||
self.assertEqual(project.layout_type, layout_type_backup)
|
||||
self.assertEqual(project.keys, keys_backup)
|
||||
|
||||
self.assertEqual(project('delegation').keys, delegation_keys_backup)
|
||||
|
||||
self.assertEqual(project._prefix, prefix_backup)
|
||||
self.assertEqual(project._project_name, name_backup)
|
||||
self.assertEqual(project.prefix, prefix_backup)
|
||||
self.assertEqual(project.project_name, name_backup)
|
||||
|
||||
roleinfo = tuf.roledb.get_roleinfo(project._project_name)
|
||||
roleinfo = tuf.roledb.get_roleinfo(project.project_name)
|
||||
|
||||
self.assertEqual(roleinfo['partial_loaded'], True)
|
||||
|
||||
|
|
@ -385,7 +385,7 @@ def test_write(self):
|
|||
|
||||
# + backup delegations.
|
||||
delegations_backup = \
|
||||
tuf.roledb.get_delegated_rolenames(project._project_name)
|
||||
tuf.roledb.get_delegated_rolenames(project.project_name)
|
||||
|
||||
# + backup layout type.
|
||||
layout_type_backup = project.layout_type
|
||||
|
|
@ -395,10 +395,10 @@ def test_write(self):
|
|||
delegation_keys_backup = project('delegation').keys
|
||||
|
||||
# + backup the prefix.
|
||||
prefix_backup = project._prefix
|
||||
prefix_backup = project.prefix
|
||||
|
||||
# + backup the name.
|
||||
name_backup = project._project_name
|
||||
name_backup = project.project_name
|
||||
|
||||
# Call status (for the sake of doing it.)
|
||||
project.status()
|
||||
|
|
@ -413,13 +413,13 @@ def test_write(self):
|
|||
# Check against backup.
|
||||
self.assertEqual(list(project.target_files.keys()), list(targets_backup.keys()))
|
||||
|
||||
new_delegations = tuf.roledb.get_delegated_rolenames(project._project_name)
|
||||
new_delegations = tuf.roledb.get_delegated_rolenames(project.project_name)
|
||||
self.assertEqual(new_delegations, delegations_backup)
|
||||
self.assertEqual(project.layout_type, layout_type_backup)
|
||||
self.assertEqual(project.keys, keys_backup)
|
||||
self.assertEqual(project('delegation').keys, delegation_keys_backup)
|
||||
self.assertEqual(project._prefix, prefix_backup)
|
||||
self.assertEqual(project._project_name, name_backup)
|
||||
self.assertEqual(project.prefix, prefix_backup)
|
||||
self.assertEqual(project.project_name, name_backup)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -560,8 +560,8 @@ def _import_delegations(self, parent_role):
|
|||
# for the key.
|
||||
try:
|
||||
key, keyids = securesystemslib.keys.format_metadata_to_key(keyinfo)
|
||||
for keyid in keyids:
|
||||
key['keyid'] = keyid
|
||||
for key_id in keyids:
|
||||
key['keyid'] = key_id
|
||||
tuf.keydb.add_key(key, keyid=None, repository_name=self.repository_name)
|
||||
|
||||
except securesystemslib.exceptions.KeyAlreadyExistsError:
|
||||
|
|
@ -1134,15 +1134,17 @@ def _get_metadata_file(self, metadata_role, remote_filename,
|
|||
|
||||
|
||||
|
||||
def _verify_root_chain_link(self, role, current, next):
|
||||
if role != 'root':
|
||||
def _verify_root_chain_link(self, rolename, current_root_metadata,
|
||||
next_root_metadata):
|
||||
|
||||
if rolename != 'root':
|
||||
return True
|
||||
|
||||
current_role = current['roles'][role]
|
||||
current_root_role = current_root_metadata['roles'][rolename]
|
||||
|
||||
# Verify next metadata with current keys/threshold
|
||||
valid = tuf.sig.verify(next, role, self.repository_name,
|
||||
current_role['threshold'], current_role['keyids'])
|
||||
valid = tuf.sig.verify(next_root_metadata, rolename, self.repository_name,
|
||||
current_root_role['threshold'], current_root_role['keyids'])
|
||||
|
||||
if not valid:
|
||||
raise securesystemslib.exceptions.BadSignatureError('Root is not signed'
|
||||
|
|
@ -1235,8 +1237,8 @@ def _get_file(self, filepath, verify_file_function, file_type,
|
|||
return file_object
|
||||
|
||||
else:
|
||||
logger.error('Failed to update {0} from all mirrors: {1}'.format(
|
||||
filepath, file_mirror_errors))
|
||||
logger.error('Failed to update ' + repr(filepath) + ' from'
|
||||
' all mirrors: ' + repr(file_mirror_errors))
|
||||
raise tuf.exceptions.NoWorkingMirrorError(file_mirror_errors)
|
||||
|
||||
|
||||
|
|
@ -1316,7 +1318,7 @@ def _update_metadata(self, metadata_role, upperbound_filelength, version=None):
|
|||
# First, move the 'current' metadata file to the 'previous' directory
|
||||
# if it exists.
|
||||
current_filepath = os.path.join(self.metadata_directory['current'],
|
||||
metadata_filename)
|
||||
metadata_filename)
|
||||
current_filepath = os.path.abspath(current_filepath)
|
||||
securesystemslib.util.ensure_parent_dir(current_filepath)
|
||||
|
||||
|
|
@ -1344,7 +1346,7 @@ def _update_metadata(self, metadata_role, upperbound_filelength, version=None):
|
|||
current_metadata_object = self.metadata['current'].get(metadata_role)
|
||||
|
||||
self._verify_root_chain_link(metadata_role, current_metadata_object,
|
||||
metadata_signable)
|
||||
metadata_signable)
|
||||
|
||||
# Finally, update the metadata and fileinfo stores, and rebuild the
|
||||
# key and role info for the top-level roles if 'metadata_role' is root.
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@
|
|||
|
||||
import os
|
||||
import errno
|
||||
import sys
|
||||
import logging
|
||||
import shutil
|
||||
import tempfile
|
||||
|
|
@ -42,6 +41,7 @@
|
|||
import tuf.roledb
|
||||
import tuf.sig
|
||||
import tuf.log
|
||||
import tuf.repository_lib as repo_lib
|
||||
import tuf.repository_tool
|
||||
|
||||
import securesystemslib
|
||||
|
|
@ -50,48 +50,20 @@
|
|||
|
||||
import six
|
||||
|
||||
# These imports provide the interface for 'developer_tool.py', since the imports
|
||||
# are made there.
|
||||
# These imports provide the interface for 'developer_tool.py', since the
|
||||
# imports are made there.
|
||||
from securesystemslib.keys import format_keyval_to_metadata
|
||||
from securesystemslib.keys import format_metadata_to_key
|
||||
|
||||
from tuf.repository_tool import Targets
|
||||
from tuf.repository_lib import get_metadata_fileinfo
|
||||
from tuf.repository_lib import get_metadata_filenames
|
||||
from tuf.repository_tool import generate_and_write_rsa_keypair
|
||||
from tuf.repository_tool import import_rsa_publickey_from_file
|
||||
from tuf.repository_tool import import_rsa_privatekey_from_file
|
||||
from tuf.repository_tool import generate_and_write_ed25519_keypair
|
||||
from tuf.repository_tool import import_ed25519_publickey_from_file
|
||||
from tuf.repository_tool import import_ed25519_privatekey_from_file
|
||||
from tuf.repository_lib import _remove_invalid_and_duplicate_signatures
|
||||
from tuf.repository_lib import disable_console_log_messages
|
||||
from tuf.repository_lib import _check_role_keys
|
||||
from tuf.repository_lib import _delete_obsolete_metadata
|
||||
from tuf.repository_lib import generate_targets_metadata
|
||||
from tuf.repository_lib import sign_metadata
|
||||
from tuf.repository_lib import write_metadata_file
|
||||
from tuf.repository_lib import _metadata_is_partially_loaded
|
||||
|
||||
# See 'log.py' to learn how logging is handled in TUF.
|
||||
logger = logging.getLogger('tuf.developer_tool')
|
||||
|
||||
# Recommended RSA key sizes:
|
||||
# http://www.emc.com/emc-plus/rsa-labs/historical/twirl-and-rsa-key-size.htm#table1
|
||||
# According to the document above, revised May 6, 2003, RSA keys of size 3072
|
||||
# provide security through 2031 and beyond. 2048-bit keys are the recommended
|
||||
# minimum and are good from the present through 2030.
|
||||
from tuf.repository_lib import DEFAULT_RSA_KEY_BITS as DEFAULT_RSA_KEY_BITS
|
||||
|
||||
# The algorithm used by the developer tools to generate the hashes of the
|
||||
# target filepaths.
|
||||
from tuf.repository_tool import HASH_FUNCTION as HASH_FUNCTION
|
||||
|
||||
# The extension of TUF metadata.
|
||||
from tuf.repository_lib import METADATA_EXTENSION as METADATA_EXTENSION
|
||||
|
||||
# The metadata filename for the targets metadata information.
|
||||
from tuf.repository_lib import TARGETS_FILENAME as TARGETS_FILENAME
|
||||
from tuf.repository_lib import METADATA_EXTENSION as METADATA_EXTENSION
|
||||
|
||||
# Project configuration filename. This file is intended to hold all of the
|
||||
# supporting information about the project that's not contained in a usual
|
||||
|
|
@ -127,16 +99,9 @@
|
|||
|
||||
# The targets and metadata directory names. Metadata files are written
|
||||
# to the staged metadata directory instead of the "live" one.
|
||||
from tuf.repository_tool import METADATA_STAGED_DIRECTORY_NAME
|
||||
from tuf.repository_tool import METADATA_DIRECTORY_NAME
|
||||
from tuf.repository_tool import TARGETS_DIRECTORY_NAME
|
||||
|
||||
# The full list of supported TUF metadata extensions.
|
||||
from tuf.repository_lib import METADATA_EXTENSIONS
|
||||
|
||||
# Supported key types.
|
||||
from tuf.repository_lib import SUPPORTED_KEY_TYPES
|
||||
|
||||
|
||||
class Project(Targets):
|
||||
"""
|
||||
|
|
@ -144,10 +109,10 @@ class Project(Targets):
|
|||
Simplify the publishing process of third-party projects by handling all of
|
||||
the bookkeeping, signature handling, and integrity checks of delegated TUF
|
||||
metadata. 'repository_tool.py' is responsible for publishing and
|
||||
maintaining metadata of the top-level roles, and 'developer_tool.py' is used
|
||||
by projects that have been delegated responsibility for a delegated projects
|
||||
role. Metadata created by this module may then be added to other metadata
|
||||
available in a TUF repository.
|
||||
maintaining metadata of the top-level roles, and 'developer_tool.py' is
|
||||
used by projects that have been delegated responsibility for a delegated
|
||||
projects role. Metadata created by this module may then be added to other
|
||||
metadata available in a TUF repository.
|
||||
|
||||
Project() is the representation of a project's metadata file(s), with the
|
||||
ability to modify this data in an OOP manner. Project owners do not have to
|
||||
|
|
@ -205,11 +170,11 @@ def __init__(self, project_name, metadata_directory, targets_directory,
|
|||
securesystemslib.formats.PATH_SCHEMA.check_match(file_prefix)
|
||||
securesystemslib.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
self._metadata_directory = metadata_directory
|
||||
self._targets_directory = targets_directory
|
||||
self._project_name = project_name
|
||||
self._prefix = file_prefix
|
||||
self._repository_name = repository_name
|
||||
self.metadata_directory = metadata_directory
|
||||
self.targets_directory = targets_directory
|
||||
self.project_name = project_name
|
||||
self.prefix = file_prefix
|
||||
self.repository_name = repository_name
|
||||
|
||||
# Layout type defaults to "flat" unless explicitly specified in
|
||||
# create_new_project().
|
||||
|
|
@ -217,7 +182,7 @@ def __init__(self, project_name, metadata_directory, targets_directory,
|
|||
|
||||
# Set the top-level Targets object. Set the rolename to be the project's
|
||||
# name.
|
||||
super(Project, self).__init__(self._targets_directory, project_name)
|
||||
super(Project, self).__init__(self.targets_directory, project_name)
|
||||
|
||||
|
||||
|
||||
|
|
@ -261,11 +226,11 @@ def write(self, write_partial=False):
|
|||
# any of the project roles are missing signatures, keys, etc.
|
||||
|
||||
# Write the metadata files of all the delegated roles of the project.
|
||||
delegated_rolenames = tuf.roledb.get_delegated_rolenames(self._project_name,
|
||||
self._repository_name)
|
||||
delegated_rolenames = tuf.roledb.get_delegated_rolenames(self.project_name,
|
||||
self.repository_name)
|
||||
|
||||
for delegated_rolename in delegated_rolenames:
|
||||
delegated_filename = os.path.join(self._metadata_directory,
|
||||
delegated_filename = os.path.join(self.metadata_directory,
|
||||
delegated_rolename + METADATA_EXTENSION)
|
||||
|
||||
# Ensure the parent directories of 'metadata_filepath' exist, otherwise an
|
||||
|
|
@ -274,29 +239,28 @@ def write(self, write_partial=False):
|
|||
securesystemslib.util.ensure_parent_dir(delegated_filename)
|
||||
|
||||
_generate_and_write_metadata(delegated_rolename, delegated_filename,
|
||||
write_partial, self._targets_directory, self._metadata_directory,
|
||||
prefix=self._prefix, repository_name=self._repository_name)
|
||||
write_partial, self.targets_directory, prefix=self.prefix,
|
||||
repository_name=self.repository_name)
|
||||
|
||||
|
||||
# Generate the 'project_name' metadata file.
|
||||
targets_filename = self._project_name + METADATA_EXTENSION
|
||||
targets_filename = os.path.join(self._metadata_directory, targets_filename)
|
||||
project_signable, targets_filename = _generate_and_write_metadata(self._project_name,
|
||||
targets_filename, write_partial, self._targets_directory,
|
||||
self._metadata_directory, prefix=self._prefix,
|
||||
repository_name=self._repository_name)
|
||||
targets_filename = self.project_name + METADATA_EXTENSION
|
||||
targets_filename = os.path.join(self.metadata_directory, targets_filename)
|
||||
junk, targets_filename = _generate_and_write_metadata(self.project_name,
|
||||
targets_filename, write_partial, self.targets_directory,
|
||||
prefix=self.prefix, repository_name=self.repository_name)
|
||||
|
||||
# Save configuration information that is not stored in the project's
|
||||
# metadata
|
||||
_save_project_configuration(self._metadata_directory,
|
||||
self._targets_directory, self.keys, self._prefix, self.threshold,
|
||||
self.layout_type, self._project_name)
|
||||
_save_project_configuration(self.metadata_directory,
|
||||
self.targets_directory, self.keys, self.prefix, self.threshold,
|
||||
self.layout_type, self.project_name)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def add_verification_key(self, key):
|
||||
def add_verification_key(self, key, expires=None):
|
||||
"""
|
||||
<Purpose>
|
||||
Function as a thin wrapper call for the project._targets call
|
||||
|
|
@ -330,7 +294,7 @@ def add_verification_key(self, key):
|
|||
raise securesystemslib.exceptions.Error("This project already contains a key.")
|
||||
|
||||
try:
|
||||
super(Project, self).add_verification_key(key)
|
||||
super(Project, self).add_verification_key(key, expires)
|
||||
|
||||
except securesystemslib.exceptions.FormatError:
|
||||
raise
|
||||
|
|
@ -369,36 +333,35 @@ def status(self):
|
|||
try:
|
||||
temp_project_directory = tempfile.mkdtemp()
|
||||
metadata_directory = os.path.join(temp_project_directory,
|
||||
self._metadata_directory[1:])
|
||||
self.metadata_directory[1:])
|
||||
|
||||
targets_directory = self._targets_directory
|
||||
targets_directory = self.targets_directory
|
||||
|
||||
os.makedirs(metadata_directory)
|
||||
|
||||
# TODO: We should do the schema check.
|
||||
filenames = {}
|
||||
filenames['targets'] = os.path.join(metadata_directory, self._project_name)
|
||||
filenames['targets'] = os.path.join(metadata_directory, self.project_name)
|
||||
|
||||
# Delegated roles.
|
||||
delegated_roles = tuf.roledb.get_delegated_rolenames(self._project_name,
|
||||
self._repository_name)
|
||||
delegated_roles = tuf.roledb.get_delegated_rolenames(self.project_name,
|
||||
self.repository_name)
|
||||
insufficient_keys = []
|
||||
insufficient_signatures = []
|
||||
|
||||
for delegated_role in delegated_roles:
|
||||
try:
|
||||
_check_role_keys(delegated_role, self._repository_name)
|
||||
_check_role_keys(delegated_role, self.repository_name)
|
||||
|
||||
except securesystemslib.exceptions.InsufficientKeysError:
|
||||
insufficient_keys.append(delegated_role)
|
||||
continue
|
||||
|
||||
roleinfo = tuf.roledb.get_roleinfo(delegated_role, self._repository_name)
|
||||
try:
|
||||
signable = _generate_and_write_metadata(delegated_role,
|
||||
filenames['targets'], False, targets_directory, metadata_directory,
|
||||
False, repository_name=self._repository_name)
|
||||
self._log_status(delegated_role, signable[0], self._repository_name)
|
||||
signable = _generate_and_write_metadata(delegated_role,
|
||||
filenames['targets'], False, targets_directory, False,
|
||||
repository_name=self.repository_name)
|
||||
self._log_status(delegated_role, signable[0], self.repository_name)
|
||||
|
||||
except securesystemslib.exceptions.Error:
|
||||
insufficient_signatures.append(delegated_role)
|
||||
|
|
@ -417,21 +380,21 @@ def status(self):
|
|||
|
||||
# Targets role.
|
||||
try:
|
||||
_check_role_keys(self.rolename, self._repository_name)
|
||||
_check_role_keys(self.rolename, self.repository_name)
|
||||
|
||||
except securesystemslib.exceptions.InsufficientKeysError as e:
|
||||
logger.info(str(e))
|
||||
return
|
||||
|
||||
try:
|
||||
signable, junk = _generate_and_write_metadata(self._project_name,
|
||||
signable, junk = _generate_and_write_metadata(self.project_name,
|
||||
filenames['targets'], False, targets_directory, metadata_directory,
|
||||
False, self._repository_name)
|
||||
self._log_status(self._project_name, signable, self._repository_name)
|
||||
self.repository_name)
|
||||
self._log_status(self.project_name, signable, self.repository_name)
|
||||
|
||||
except securesystemslib.exceptions.Error as e:
|
||||
signable = e[1]
|
||||
self._log_status(self._project_name, signable, self._repository_name)
|
||||
self._log_status(self.project_name, signable, self.repository_name)
|
||||
return
|
||||
|
||||
finally:
|
||||
|
|
@ -459,8 +422,7 @@ def _log_status(self, rolename, signable, repository_name):
|
|||
|
||||
|
||||
def _generate_and_write_metadata(rolename, metadata_filename, write_partial,
|
||||
targets_directory, metadata_directory, filenames=None, prefix='',
|
||||
repository_name='default'):
|
||||
targets_directory, prefix='', repository_name='default'):
|
||||
"""
|
||||
Non-public function that can generate and write the metadata of the
|
||||
specified 'rolename'. It also increments version numbers if:
|
||||
|
|
@ -483,15 +445,14 @@ def _generate_and_write_metadata(rolename, metadata_filename, write_partial,
|
|||
|
||||
# Prepend the prefix to the project's filepath to avoid signature errors in
|
||||
# upstream.
|
||||
target_filepaths = metadata['targets'].items()
|
||||
for element in list(metadata['targets']):
|
||||
junk_path, relative_target = os.path.split(element)
|
||||
prefixed_path = os.path.join(prefix,relative_target)
|
||||
junk, relative_target = os.path.split(element)
|
||||
prefixed_path = os.path.join(prefix, relative_target)
|
||||
metadata['targets'][prefixed_path] = metadata['targets'][element]
|
||||
if prefix != '':
|
||||
del(metadata['targets'][element])
|
||||
|
||||
signable = sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
signable = repo_lib.sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
metadata_filename, repository_name)
|
||||
|
||||
# Check if the version number of 'rolename' may be automatically incremented,
|
||||
|
|
@ -499,29 +460,29 @@ def _generate_and_write_metadata(rolename, metadata_filename, write_partial,
|
|||
# written with write() / write_partial().
|
||||
# Increment the version number if this is the first partial write.
|
||||
if write_partial:
|
||||
temp_signable = sign_metadata(metadata, [], metadata_filename,
|
||||
temp_signable = repo_lib.sign_metadata(metadata, [], metadata_filename,
|
||||
repository_name)
|
||||
temp_signable['signatures'].extend(roleinfo['signatures'])
|
||||
status = tuf.sig.get_signature_status(temp_signable, rolename,
|
||||
repository_name)
|
||||
if len(status['good_sigs']) == 0:
|
||||
metadata['version'] = metadata['version'] + 1
|
||||
signable = sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
signable = repo_lib.sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
metadata_filename, repository_name)
|
||||
|
||||
# non-partial write()
|
||||
else:
|
||||
if tuf.sig.verify(signable, rolename, repository_name):
|
||||
metadata['version'] = metadata['version'] + 1
|
||||
signable = sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
signable = repo_lib.sign_metadata(metadata, roleinfo['signing_keyids'],
|
||||
metadata_filename, repository_name)
|
||||
|
||||
# Write the metadata to file if contains a threshold of signatures.
|
||||
signable['signatures'].extend(roleinfo['signatures'])
|
||||
|
||||
if tuf.sig.verify(signable, rolename, repository_name) or write_partial:
|
||||
_remove_invalid_and_duplicate_signatures(signable, repository_name)
|
||||
filename = write_metadata_file(signable, metadata_filename,
|
||||
repo_lib._remove_invalid_and_duplicate_signatures(signable, repository_name)
|
||||
filename = repo_lib.write_metadata_file(signable, metadata_filename,
|
||||
metadata['version'], False)
|
||||
|
||||
# 'signable' contains an invalid threshold of signatures.
|
||||
|
|
@ -738,9 +699,9 @@ def _save_project_configuration(metadata_directory, targets_directory,
|
|||
# If it is, the .cfg file should be saved in the previous directory.
|
||||
if layout_type == 'repo-like':
|
||||
cfg_file_directory = os.path.dirname(metadata_directory)
|
||||
absolute_location, targets_directory = os.path.split(targets_directory)
|
||||
junk, targets_directory = os.path.split(targets_directory)
|
||||
|
||||
absolute_location, metadata_directory = os.path.split(metadata_directory)
|
||||
junk, metadata_directory = os.path.split(metadata_directory)
|
||||
|
||||
# Can the file be opened?
|
||||
project_filename = os.path.join(cfg_file_directory, PROJECT_FILENAME)
|
||||
|
|
@ -860,7 +821,7 @@ def load_project(project_directory, prefix='', new_targets_location=None,
|
|||
repository_name)
|
||||
|
||||
project.threshold = project_configuration['threshold']
|
||||
project._prefix = project_configuration['prefix']
|
||||
project.prefix = project_configuration['prefix']
|
||||
project.layout_type = project_configuration['layout_type']
|
||||
|
||||
# Traverse the public keys and add them to the project.
|
||||
|
|
@ -1002,7 +963,7 @@ def load_project(project_directory, prefix='', new_targets_location=None,
|
|||
tuf.roledb.add_role(rolename, roleinfo, repository_name=repository_name)
|
||||
|
||||
if new_prefix:
|
||||
project._prefix = new_prefix
|
||||
project.prefix = new_prefix
|
||||
|
||||
return project
|
||||
|
||||
|
|
@ -1031,6 +992,35 @@ def _strip_prefix_from_targets_metadata(targets_metadata, prefix):
|
|||
|
||||
|
||||
|
||||
# Wrapper functions that we wish to make available here from repository_lib.py.
|
||||
# Users are expected to call functions provided by repository_tool.py. We opt
|
||||
# for this approach, as opposed to using import statements to achieve the
|
||||
# equivalent, to avoid linter warnings for unused imports.
|
||||
def generate_and_write_rsa_keypair(filepath, bits, password):
|
||||
return repo_lib.generate_and_write_rsa_keypair(filepath, bits, password)
|
||||
|
||||
def generate_and_write_ed25519_keypair(filepath, password):
|
||||
return repo_lib.generate_and_write_ed25519_keypair(filepath, password)
|
||||
|
||||
def import_rsa_publickey_from_file(filepath):
|
||||
return repo_lib.import_rsa_publickey_from_file(filepath)
|
||||
|
||||
def import_ed25519_publickey_from_file(filepath):
|
||||
return repo_lib.import_ed25519_publickey_from_file(filepath)
|
||||
|
||||
def import_rsa_privatekey_from_file(filepath, password):
|
||||
return repo_lib.import_rsa_privatekey_from_file(filepath, password)
|
||||
|
||||
def import_ed25519_privatekey_from_file(filepath, password):
|
||||
return repo_lib.import_ed25519_privatekey_from_file(filepath, password)
|
||||
|
||||
def create_tuf_client_directory(repository_directory, client_directory):
|
||||
return repo_lib.create_tuf_client_directory(repository_directory, client_directory)
|
||||
|
||||
def disable_console_log_messages():
|
||||
return repo_lib.disable_console_log_messages()
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -530,10 +530,9 @@ def _get_content_length(connection):
|
|||
except Exception as e:
|
||||
logger.exception('Could not get content length'
|
||||
' about ' + str(connection) + ' from server: ' + str(e))
|
||||
reported_length = None
|
||||
return None
|
||||
|
||||
finally:
|
||||
return reported_length
|
||||
return reported_length
|
||||
|
||||
|
||||
|
||||
|
|
@ -692,14 +691,14 @@ class VerifiedHTTPSConnection(six.moves.http_client.HTTPSConnection):
|
|||
|
||||
def connect(self):
|
||||
|
||||
self.connection_kwargs = {}
|
||||
self.connection_kwargs.update(timeout = self.timeout)
|
||||
connection_kwargs = {}
|
||||
connection_kwargs.update(timeout = self.timeout)
|
||||
|
||||
# for >= py2.7
|
||||
if hasattr(self, 'source_address'):
|
||||
self.connection_kwargs.update(source_address = self.source_address)
|
||||
connection_kwargs.update(source_address = self.source_address)
|
||||
|
||||
sock = socket.create_connection((self.host, self.port), **self.connection_kwargs)
|
||||
sock = socket.create_connection((self.host, self.port), **connection_kwargs)
|
||||
|
||||
# for >= py2.7
|
||||
if getattr(self, '_tunnel_host', None):
|
||||
|
|
@ -722,7 +721,8 @@ def connect(self):
|
|||
# ssl.PROTOCOL_SSLv23, the default value for 'ssl_version', is deprecated in
|
||||
# Python 2.7.13, but becomes an alias for ssl.PROTOCOL_TLS.
|
||||
self.sock = ssl.wrap_socket(sock, self.key_file, self.cert_file,
|
||||
cert_reqs=ssl.CERT_REQUIRED, ca_certs=cert_path, ssl_version=ssl.PROTOCOL_SSLv23)
|
||||
cert_reqs=ssl.CERT_REQUIRED, ca_certs=cert_path,
|
||||
ssl_version=ssl.PROTOCOL_SSLv23)
|
||||
|
||||
match_hostname(self.sock.getpeercert(), self.host)
|
||||
|
||||
|
|
|
|||
|
|
@ -202,7 +202,7 @@ def remove_keydb(repository_name):
|
|||
securesystemslib.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
if repository_name not in _keydb_dict:
|
||||
logger.warn('Repository name does not exist: ' + repr(repository_name))
|
||||
logger.warning('Repository name does not exist: ' + repr(repository_name))
|
||||
return
|
||||
|
||||
if repository_name == 'default':
|
||||
|
|
|
|||
|
|
@ -166,7 +166,7 @@ def filter(self, record):
|
|||
# file logging handler, the user may always consult the file log for the
|
||||
# original exception traceback. The exc_info is explained here:
|
||||
# http://docs.python.org/2/library/sys.html#sys.exc_info
|
||||
exc_type, exc_value, exc_traceback = record.exc_info
|
||||
exc_type, junk, junk = record.exc_info
|
||||
|
||||
# Simply set the class name as the exception text.
|
||||
record.exc_text = exc_type.__name__
|
||||
|
|
@ -316,10 +316,10 @@ def add_console_handler(log_level=_DEFAULT_CONSOLE_LOG_LEVEL):
|
|||
# Get our filter for the console handler.
|
||||
console_filter = ConsoleFilter()
|
||||
console_format_string = '%(message)s'
|
||||
formatter = logging.Formatter(console_format_string)
|
||||
console_formatter = logging.Formatter(console_format_string)
|
||||
|
||||
console_handler.setLevel(log_level)
|
||||
console_handler.setFormatter(formatter)
|
||||
console_handler.setFormatter(console_formatter)
|
||||
console_handler.addFilter(console_filter)
|
||||
logger.addHandler(console_handler)
|
||||
logger.debug('Added a console handler.')
|
||||
|
|
|
|||
|
|
@ -563,9 +563,6 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
logger.debug('Found a Root signature that is already loaded:'
|
||||
' ' + repr(signature))
|
||||
|
||||
else:
|
||||
logger.debug('A compressed Root file was not found.')
|
||||
|
||||
# By default, roleinfo['partial_loaded'] of top-level roles should be set
|
||||
# to False in 'create_roledb_from_root_metadata()'. Update this field, if
|
||||
# necessary, now that we have its signable object.
|
||||
|
|
@ -1747,8 +1744,7 @@ def write_metadata_file(metadata, filename, version_number, consistent_snapshot)
|
|||
Any other runtime (e.g., IO) exception.
|
||||
|
||||
<Side Effects>
|
||||
The 'filename' (or the compressed filename) file is created, or overwritten
|
||||
if it exists.
|
||||
The 'filename' file is created, or overwritten if it exists.
|
||||
|
||||
<Returns>
|
||||
The filename of the written file.
|
||||
|
|
@ -1779,10 +1775,9 @@ def write_metadata_file(metadata, filename, version_number, consistent_snapshot)
|
|||
# not been previously written or has changed). It is now assumed that the
|
||||
# caller intends to write changes that have been marked as dirty.
|
||||
|
||||
# The 'metadata' object is written to 'file_object', including compressed
|
||||
# versions. To avoid partial metadata from being written, 'metadata' is
|
||||
# first written to a temporary location (i.e., 'file_object') and then
|
||||
# moved to 'filename'.
|
||||
# The 'metadata' object is written to 'file_object'. To avoid partial
|
||||
# metadata from being written, 'metadata' is first written to a temporary
|
||||
# location (i.e., 'file_object') and then moved to 'filename'.
|
||||
file_object = securesystemslib.util.TempFile()
|
||||
|
||||
# Serialize 'metadata' to the file-like object and then write
|
||||
|
|
|
|||
|
|
@ -223,7 +223,7 @@ def remove_roledb(repository_name):
|
|||
global _dirty_roles
|
||||
|
||||
if repository_name not in _roledb_dict or repository_name not in _dirty_roles:
|
||||
logger.warn('Repository name does not exist:'
|
||||
logger.warning('Repository name does not exist:'
|
||||
' ' + repr(repository_name))
|
||||
return
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue