mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
Jussi Kukkonen
GitHub
commit
08f48d52df
33 changed files with 99 additions and 127 deletions
4
pylintrc
4
pylintrc
|
|
@ -287,7 +287,7 @@ ignore-on-opaque-inference=yes
|
|||
# List of class names for which member attributes should not be checked (useful
|
||||
# for classes with dynamically set attributes). This supports the use of
|
||||
# qualified names.
|
||||
ignored-classes=optparse.Values,thread._local,_thread._local, six.moves
|
||||
ignored-classes=optparse.Values,thread._local,_thread._local
|
||||
|
||||
# List of module names for which member attributes should not be checked
|
||||
# (useful for modules/projects where namespaces are manipulated during runtime
|
||||
|
|
@ -334,7 +334,7 @@ init-import=no
|
|||
|
||||
# List of qualified module names which can have objects that can redefine
|
||||
# builtins.
|
||||
redefining-builtins-modules=six.moves,future.builtins
|
||||
redefining-builtins-modules=future.builtins
|
||||
|
||||
|
||||
[CLASSES]
|
||||
|
|
|
|||
|
|
@ -7,5 +7,4 @@ pycparser==2.20 # via cffi
|
|||
pynacl==1.4.0 # via securesystemslib
|
||||
requests==2.25.1
|
||||
securesystemslib[crypto,pynacl]==0.20.0
|
||||
six==1.15.0
|
||||
urllib3==1.26.4 # via requests
|
||||
|
|
|
|||
|
|
@ -43,4 +43,3 @@
|
|||
#
|
||||
securesystemslib[crypto, pynacl]
|
||||
requests
|
||||
six
|
||||
|
|
|
|||
3
setup.py
3
setup.py
|
|
@ -113,8 +113,7 @@
|
|||
python_requires="~=3.6",
|
||||
install_requires = [
|
||||
'requests>=2.19.1',
|
||||
'securesystemslib>=0.20.0',
|
||||
'six>=1.11.0'
|
||||
'securesystemslib>=0.20.0'
|
||||
],
|
||||
packages = find_packages(exclude=['tests']),
|
||||
scripts = [
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@
|
|||
import sys
|
||||
import ssl
|
||||
import os
|
||||
import six
|
||||
import http.server
|
||||
|
||||
keyfile = os.path.join('ssl_certs', 'ssl_cert.key')
|
||||
certfile = os.path.join('ssl_certs', 'ssl_cert.crt')
|
||||
|
|
@ -49,8 +49,8 @@
|
|||
if len(sys.argv) > 1 and os.path.exists(sys.argv[1]):
|
||||
certfile = sys.argv[1]
|
||||
|
||||
httpd = six.moves.BaseHTTPServer.HTTPServer(('localhost', 0),
|
||||
six.moves.SimpleHTTPServer.SimpleHTTPRequestHandler)
|
||||
httpd = http.server.HTTPServer(('localhost', 0),
|
||||
http.server.SimpleHTTPRequestHandler)
|
||||
|
||||
httpd.socket = ssl.wrap_socket(
|
||||
httpd.socket, keyfile=keyfile, certfile=certfile, server_side=True)
|
||||
|
|
|
|||
|
|
@ -35,9 +35,8 @@
|
|||
|
||||
import sys
|
||||
import random
|
||||
|
||||
import six
|
||||
from six.moves.SimpleHTTPServer import SimpleHTTPRequestHandler
|
||||
import socketserver
|
||||
from http.server import SimpleHTTPRequestHandler
|
||||
|
||||
|
||||
class QuietHTTPRequestHandler(SimpleHTTPRequestHandler):
|
||||
|
|
@ -64,9 +63,9 @@ def log_request(self, code='-', size='-'):
|
|||
|
||||
# Allow re-use so you can re-run tests as often as you want even if the
|
||||
# tests re-use ports. Otherwise TCP TIME-WAIT prevents reuse for ~1 minute
|
||||
six.moves.socketserver.TCPServer.allow_reuse_address = True
|
||||
socketserver.TCPServer.allow_reuse_address = True
|
||||
|
||||
httpd = six.moves.socketserver.TCPServer(('localhost', 0), handler)
|
||||
httpd = socketserver.TCPServer(('localhost', 0), handler)
|
||||
port_message = 'bind succeeded, server port is: ' \
|
||||
+ str(httpd.server_address[1])
|
||||
print(port_message)
|
||||
|
|
|
|||
|
|
@ -32,13 +32,12 @@
|
|||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import six
|
||||
import http.server
|
||||
|
||||
|
||||
|
||||
# HTTP request handler.
|
||||
class Handler(six.moves.BaseHTTPServer.BaseHTTPRequestHandler):
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
# Overwrite do_GET.
|
||||
def do_GET(self):
|
||||
|
|
@ -66,7 +65,7 @@ def do_GET(self):
|
|||
if __name__ == '__main__':
|
||||
server_address = ('localhost', 0)
|
||||
|
||||
httpd = six.moves.BaseHTTPServer.HTTPServer(server_address, Handler)
|
||||
httpd = http.server.HTTPServer(server_address, Handler)
|
||||
port_message = 'bind succeeded, server port is: ' \
|
||||
+ str(httpd.server_address[1])
|
||||
print(port_message)
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@
|
|||
import logging
|
||||
import unittest
|
||||
import sys
|
||||
from urllib import request
|
||||
|
||||
import tuf
|
||||
import tuf.formats
|
||||
|
|
@ -55,7 +56,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -174,7 +174,7 @@ def test_without_tuf(self):
|
|||
url_file = os.path.join(url_prefix, 'targets', 'file1.txt')
|
||||
|
||||
# On Windows, the URL portion should not contain back slashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
|
||||
self.assertTrue(os.path.exists(client_target_path))
|
||||
length, hashes = securesystemslib.util.get_file_details(client_target_path)
|
||||
|
|
@ -188,7 +188,7 @@ def test_without_tuf(self):
|
|||
malicious_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
||||
# On Windows, the URL portion should not contain back slashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
|
||||
length, hashes = securesystemslib.util.get_file_details(client_target_path)
|
||||
download_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@
|
|||
import logging
|
||||
import unittest
|
||||
import sys
|
||||
from urllib import request
|
||||
|
||||
import tuf
|
||||
import tuf.formats
|
||||
|
|
@ -57,7 +58,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -176,7 +176,7 @@ def test_without_tuf(self):
|
|||
url_file = os.path.join(url_prefix, 'targets', 'file1.txt')
|
||||
|
||||
# On Windows, the URL portion should not contain backslashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
|
||||
self.assertTrue(os.path.exists(client_target_path))
|
||||
length, hashes = securesystemslib.util.get_file_details(client_target_path)
|
||||
|
|
@ -194,7 +194,7 @@ def test_without_tuf(self):
|
|||
self.assertTrue(large_length > length)
|
||||
|
||||
# On Windows, the URL portion should not contain backslashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_target_path)
|
||||
|
||||
length, hashes = securesystemslib.util.get_file_details(client_target_path)
|
||||
download_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
|
@ -269,7 +269,7 @@ def test_with_tuf(self):
|
|||
self.repository_updater.refresh()
|
||||
|
||||
except tuf.exceptions.NoWorkingMirrorError as exception:
|
||||
for mirror_url, mirror_error in six.iteritems(exception.mirror_errors):
|
||||
for mirror_url, mirror_error in exception.mirror_errors.items():
|
||||
self.assertTrue(isinstance(mirror_error, securesystemslib.exceptions.Error))
|
||||
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -60,7 +60,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -206,7 +205,7 @@ def test_with_tuf(self):
|
|||
# Verify that the specific 'tuf.exceptions.ForbiddenTargetError' exception is raised
|
||||
# by each mirror.
|
||||
except tuf.exceptions.NoWorkingMirrorError as exception:
|
||||
for mirror_url, mirror_error in six.iteritems(exception.mirror_errors):
|
||||
for mirror_url, mirror_error in exception.mirror_errors.items():
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'metadata', 'role1.json')
|
||||
|
||||
|
|
|
|||
|
|
@ -40,7 +40,6 @@
|
|||
|
||||
import securesystemslib
|
||||
import securesystemslib.util
|
||||
import six
|
||||
|
||||
|
||||
class TestFormats(unittest.TestCase):
|
||||
|
|
@ -287,7 +286,7 @@ def test_schemas(self):
|
|||
|
||||
# Iterate 'valid_schemas', ensuring each 'valid_schema' correctly matches
|
||||
# its respective 'schema_type'.
|
||||
for schema_name, (schema_type, valid_schema) in six.iteritems(valid_schemas):
|
||||
for schema_name, (schema_type, valid_schema) in valid_schemas.items():
|
||||
if not schema_type.matches(valid_schema):
|
||||
print('bad schema: ' + repr(valid_schema))
|
||||
self.assertEqual(True, schema_type.matches(valid_schema))
|
||||
|
|
@ -295,7 +294,7 @@ def test_schemas(self):
|
|||
# Test conditions for invalid schemas.
|
||||
# Set the 'valid_schema' of 'valid_schemas' to an invalid
|
||||
# value and test that it does not match 'schema_type'.
|
||||
for schema_name, (schema_type, valid_schema) in six.iteritems(valid_schemas):
|
||||
for schema_name, (schema_type, valid_schema) in valid_schemas.items():
|
||||
invalid_schema = 0xBAD
|
||||
if isinstance(schema_type, securesystemslib.schema.Integer):
|
||||
invalid_schema = 'BAD'
|
||||
|
|
@ -779,7 +778,7 @@ def test_format_base64(self):
|
|||
# Test conditions for valid arguments.
|
||||
data = 'updateframework'.encode('utf-8')
|
||||
self.assertEqual('dXBkYXRlZnJhbWV3b3Jr', tuf.formats.format_base64(data))
|
||||
self.assertTrue(isinstance(tuf.formats.format_base64(data), six.string_types))
|
||||
self.assertTrue(isinstance(tuf.formats.format_base64(data), str))
|
||||
|
||||
# Test conditions for invalid arguments.
|
||||
self.assertRaises(securesystemslib.exceptions.FormatError, tuf.formats.format_base64, 123)
|
||||
|
|
@ -791,7 +790,7 @@ def test_parse_base64(self):
|
|||
# Test conditions for valid arguments.
|
||||
base64 = 'dXBkYXRlZnJhbWV3b3Jr'
|
||||
self.assertEqual(b'updateframework', tuf.formats.parse_base64(base64))
|
||||
self.assertTrue(isinstance(tuf.formats.parse_base64(base64), six.binary_type))
|
||||
self.assertTrue(isinstance(tuf.formats.parse_base64(base64), bytes))
|
||||
|
||||
# Test conditions for invalid arguments.
|
||||
self.assertRaises(securesystemslib.exceptions.FormatError, tuf.formats.parse_base64, 123)
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@
|
|||
import logging
|
||||
import unittest
|
||||
import sys
|
||||
from urllib import request
|
||||
|
||||
if sys.version_info >= (3, 3):
|
||||
import unittest.mock as mock
|
||||
|
|
@ -71,7 +72,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
# The repository tool is imported and logs console messages by default. Disable
|
||||
# console log messages generated by this unit test.
|
||||
|
|
@ -223,7 +223,7 @@ def test_without_tuf(self):
|
|||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'metadata', 'timestamp.json')
|
||||
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
|
||||
length, hashes = securesystemslib.util.get_file_details(client_timestamp_path)
|
||||
download_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
|
@ -367,7 +367,7 @@ def test_with_tuf(self):
|
|||
|
||||
except tuf.exceptions.NoWorkingMirrorError as e:
|
||||
# Make sure the contained error is ExpiredMetadataError
|
||||
for mirror_url, mirror_error in six.iteritems(e.mirror_errors):
|
||||
for mirror_url, mirror_error in e.mirror_errors.items():
|
||||
self.assertTrue(isinstance(mirror_error, tuf.exceptions.ExpiredMetadataError))
|
||||
|
||||
else:
|
||||
|
|
@ -427,7 +427,7 @@ def test_with_tuf(self):
|
|||
|
||||
except tuf.exceptions.NoWorkingMirrorError as e:
|
||||
# Make sure the contained error is ExpiredMetadataError
|
||||
for mirror_url, mirror_error in six.iteritems(e.mirror_errors):
|
||||
for mirror_url, mirror_error in e.mirror_errors.items():
|
||||
self.assertTrue(isinstance(mirror_error, tuf.exceptions.ExpiredMetadataError))
|
||||
self.assertTrue(mirror_url.endswith('snapshot.json'))
|
||||
|
||||
|
|
|
|||
|
|
@ -55,7 +55,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
repo_tool.disable_console_log_messages()
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@
|
|||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import importlib
|
||||
|
||||
import tuf
|
||||
import tuf.log
|
||||
|
|
@ -35,7 +36,6 @@
|
|||
|
||||
from tests import utils
|
||||
|
||||
from six.moves import reload_module
|
||||
|
||||
# We explicitly create a logger which is a child of the tuf hierarchy,
|
||||
# instead of using the standard getLogger(__name__) pattern, because the
|
||||
|
|
@ -96,7 +96,7 @@ def test_set_filehandler_log_level(self):
|
|||
# Test that the log level of the file handler cannot be set because
|
||||
# file logging is disabled (via tuf.settings.ENABLE_FILE_LOGGING).
|
||||
tuf.settings.ENABLE_FILE_LOGGING = False
|
||||
reload_module(tuf.log)
|
||||
importlib.reload(tuf.log)
|
||||
|
||||
# Test for improperly formatted argument.
|
||||
self.assertRaises(securesystemslib.exceptions.FormatError, tuf.log.set_filehandler_log_level, '123')
|
||||
|
|
|
|||
|
|
@ -55,7 +55,6 @@
|
|||
|
||||
from tests import utils
|
||||
|
||||
import six
|
||||
|
||||
# The repository tool is imported and logs console messages by default.
|
||||
# Disable console log messages generated by this unit test.
|
||||
|
|
@ -227,7 +226,7 @@ def test_with_tuf(self):
|
|||
# 'tuf.exceptions.BadVersionNumberError' exception is raised by
|
||||
# each mirror.
|
||||
except tuf.exceptions.NoWorkingMirrorError as exception:
|
||||
for mirror_url, mirror_error in six.iteritems(exception.mirror_errors):
|
||||
for mirror_url, mirror_error in exception.mirror_errors.items():
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'metadata', 'role1.json')
|
||||
|
||||
|
|
|
|||
|
|
@ -47,7 +47,6 @@
|
|||
|
||||
from tests import utils
|
||||
|
||||
import six
|
||||
import securesystemslib
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -271,7 +270,7 @@ def test_repository_tool(self):
|
|||
multi_repo_updater = updater.MultiRepoUpdater(self.map_file)
|
||||
valid_targetinfo = multi_repo_updater.get_valid_targetinfo('file3.txt')
|
||||
|
||||
for my_updater, my_targetinfo in six.iteritems(valid_targetinfo):
|
||||
for my_updater, my_targetinfo in valid_targetinfo.items():
|
||||
my_updater.download_target(my_targetinfo, self.temporary_directory)
|
||||
self.assertTrue(os.path.exists(os.path.join(self.temporary_directory, 'file3.txt')))
|
||||
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@
|
|||
import logging
|
||||
import unittest
|
||||
import sys
|
||||
from urllib import request
|
||||
|
||||
import tuf.formats
|
||||
import tuf.log
|
||||
|
|
@ -55,7 +56,7 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
|
||||
# The repository tool is imported and logs console messages by default.
|
||||
# Disable console log messages generated by this unit test.
|
||||
|
|
@ -220,7 +221,7 @@ def test_without_tuf(self):
|
|||
self.repository_name, 'metadata', 'current', 'timestamp.json')
|
||||
|
||||
# On Windows, the URL portion should not contain back slashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
|
||||
length, hashes = securesystemslib.util.get_file_details(client_timestamp_path)
|
||||
download_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
|
@ -233,7 +234,7 @@ def test_without_tuf(self):
|
|||
shutil.move(backup_timestamp, timestamp_path)
|
||||
|
||||
# On Windows, the URL portion should not contain back slashes.
|
||||
six.moves.urllib.request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
request.urlretrieve(url_file.replace('\\', '/'), client_timestamp_path)
|
||||
|
||||
length, hashes = securesystemslib.util.get_file_details(client_timestamp_path)
|
||||
download_fileinfo = tuf.formats.make_targets_fileinfo(length, hashes)
|
||||
|
|
@ -312,7 +313,7 @@ def test_with_tuf(self):
|
|||
# Verify that the specific 'tuf.exceptions.ReplayedMetadataError' is raised by each
|
||||
# mirror.
|
||||
except tuf.exceptions.NoWorkingMirrorError as exception:
|
||||
for mirror_url, mirror_error in six.iteritems(exception.mirror_errors):
|
||||
for mirror_url, mirror_error in exception.mirror_errors.items():
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'metadata', 'timestamp.json')
|
||||
|
||||
|
|
|
|||
|
|
@ -57,7 +57,6 @@
|
|||
import securesystemslib.rsa_keys
|
||||
import securesystemslib.interface
|
||||
import securesystemslib.storage
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -271,7 +270,7 @@ def test_get_target_hash(self):
|
|||
'/README.txt': '8faee106f1bb69f34aaf1df1e3c2e87d763c4d878cb96b91db13495e32ceb0b0',
|
||||
'/packages/file2.txt': 'c9c4a5cdd84858dd6a23d98d7e6e6b2aec45034946c16b2200bc317c75415e92'
|
||||
}
|
||||
for filepath, target_hash in six.iteritems(expected_target_hashes):
|
||||
for filepath, target_hash in expected_target_hashes.items():
|
||||
self.assertTrue(tuf.formats.RELPATH_SCHEMA.matches(filepath))
|
||||
self.assertTrue(securesystemslib.formats.HASH_SCHEMA.matches(target_hash))
|
||||
self.assertEqual(repo_lib.get_target_hash(filepath), target_hash)
|
||||
|
|
|
|||
|
|
@ -60,8 +60,6 @@
|
|||
|
||||
from tests import utils
|
||||
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
repo_tool.disable_console_log_messages()
|
||||
|
||||
|
|
@ -218,7 +216,7 @@ def test_delay_before_send(self):
|
|||
# Verify that the specific 'tuf.exceptions.SlowRetrievalError' exception is raised by
|
||||
# each mirror.
|
||||
except tuf.exceptions.NoWorkingMirrorError as exception:
|
||||
for mirror_url, mirror_error in six.iteritems(exception.mirror_errors):
|
||||
for mirror_url, mirror_error in exception.mirror_errors.items():
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'targets', 'file1.txt')
|
||||
|
||||
|
|
|
|||
|
|
@ -79,7 +79,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
repo_tool.disable_console_log_messages()
|
||||
|
|
@ -775,7 +774,7 @@ def test_3__update_metadata(self):
|
|||
DEFAULT_TARGETS_FILELENGTH, 88)
|
||||
|
||||
except tuf.exceptions.NoWorkingMirrorError as e:
|
||||
for mirror_error in six.itervalues(e.mirror_errors):
|
||||
for mirror_error in e.mirror_errors.values():
|
||||
assert isinstance(mirror_error, tuf.exceptions.BadVersionNumberError)
|
||||
|
||||
else:
|
||||
|
|
@ -791,7 +790,7 @@ def test_3__update_metadata(self):
|
|||
88)
|
||||
|
||||
except tuf.exceptions.NoWorkingMirrorError as e:
|
||||
for mirror_error in six.itervalues(e.mirror_errors):
|
||||
for mirror_error in e.mirror_errors.values():
|
||||
assert isinstance(mirror_error, tuf.exceptions.BadVersionNumberError)
|
||||
|
||||
else:
|
||||
|
|
@ -839,7 +838,7 @@ def test_3__get_metadata_file(self):
|
|||
# Note that this test provides a piece of metadata which would fail to
|
||||
# be accepted -- with a different error -- if the specification version
|
||||
# number were not a problem.
|
||||
for mirror_error in six.itervalues(e.mirror_errors):
|
||||
for mirror_error in e.mirror_errors.values():
|
||||
assert isinstance(
|
||||
mirror_error, tuf.exceptions.UnsupportedSpecificationError)
|
||||
|
||||
|
|
@ -921,7 +920,7 @@ def test_3__targets_of_role(self):
|
|||
# target files.
|
||||
self.assertTrue(tuf.formats.TARGETINFOS_SCHEMA.matches(targetinfos_list))
|
||||
for targetinfo in targetinfos_list:
|
||||
self.assertTrue((targetinfo['filepath'], targetinfo['fileinfo']) in six.iteritems(targets_in_metadata))
|
||||
self.assertTrue((targetinfo['filepath'], targetinfo['fileinfo']) in targets_in_metadata.items())
|
||||
|
||||
|
||||
|
||||
|
|
@ -1084,7 +1083,7 @@ def test_5_targets_of_role(self):
|
|||
# target files.
|
||||
self.assertTrue(tuf.formats.TARGETINFOS_SCHEMA.matches(targetinfos))
|
||||
for targetinfo in targetinfos:
|
||||
self.assertTrue((targetinfo['filepath'], targetinfo['fileinfo']) in six.iteritems(expected_targets))
|
||||
self.assertTrue((targetinfo['filepath'], targetinfo['fileinfo']) in expected_targets.items())
|
||||
|
||||
# Test: Invalid arguments.
|
||||
# targets_of_role() expected a string rolename.
|
||||
|
|
@ -1368,7 +1367,7 @@ def test_6_download_target(self):
|
|||
# field contains at least one confined target and excludes needed target
|
||||
# file.
|
||||
mirrors = self.repository_updater.mirrors
|
||||
for mirror_name, mirror_info in six.iteritems(mirrors):
|
||||
for mirror_name, mirror_info in mirrors.items():
|
||||
mirrors[mirror_name]['confined_target_dirs'] = [self.random_path()]
|
||||
|
||||
try:
|
||||
|
|
@ -1630,7 +1629,7 @@ def test_9__get_target_hash(self):
|
|||
'/file1.txt': 'e3a3d89eb3b70ce3fbce6017d7b8c12d4abd5635427a0e8a238f53157df85b3d',
|
||||
'/Jalape\xc3\xb1o': '78bfd5c314680545eb48ecad508aceb861f8d6e680f4fe1b791da45c298cda88'
|
||||
}
|
||||
for filepath, target_hash in six.iteritems(expected_target_hashes):
|
||||
for filepath, target_hash in expected_target_hashes.items():
|
||||
self.assertTrue(tuf.formats.RELPATH_SCHEMA.matches(filepath))
|
||||
self.assertTrue(securesystemslib.formats.HASH_SCHEMA.matches(target_hash))
|
||||
self.assertEqual(self.repository_updater._get_target_hash(filepath), target_hash)
|
||||
|
|
|
|||
|
|
@ -62,7 +62,6 @@
|
|||
from tests import utils
|
||||
|
||||
import securesystemslib
|
||||
import six
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
repo_tool.disable_console_log_messages()
|
||||
|
|
@ -248,7 +247,7 @@ def test_verify_root_with_current_keyids_and_threshold(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/2.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
@ -307,7 +306,7 @@ def test_verify_root_with_duplicate_current_keyids(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/2.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
@ -450,7 +449,7 @@ def test_root_rotation_missing_keys(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/2.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
@ -525,7 +524,7 @@ def test_root_rotation_unmet_last_version_threshold(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/3.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
@ -571,7 +570,7 @@ def test_root_rotation_unmet_new_threshold(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/3.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
@ -606,7 +605,7 @@ def test_root_rotation_discard_untrusted_version(self):
|
|||
with self.assertRaises(tuf.exceptions.NoWorkingMirrorError) as cm:
|
||||
self.repository_updater.refresh()
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(cm.exception.mirror_errors):
|
||||
for mirror_url, mirror_error in cm.exception.mirror_errors.items():
|
||||
self.assertTrue(mirror_url.endswith('/2.root.json'))
|
||||
self.assertTrue(isinstance(mirror_error,
|
||||
securesystemslib.exceptions.BadSignatureError))
|
||||
|
|
|
|||
|
|
@ -128,7 +128,6 @@
|
|||
import time
|
||||
import fnmatch
|
||||
import copy
|
||||
import six
|
||||
import warnings
|
||||
import io
|
||||
|
||||
|
|
@ -388,7 +387,7 @@ def _matching_targetinfo(
|
|||
# a threshold of 2:
|
||||
# [A, B, C, B, A, C]
|
||||
# In this case, targetinfo B is returned.
|
||||
for valid_updater, compared_targetinfo in six.iteritems(valid_targetinfo):
|
||||
for valid_updater, compared_targetinfo in valid_targetinfo.items():
|
||||
|
||||
if not self._targetinfo_match(
|
||||
targetinfo, compared_targetinfo, match_custom_field):
|
||||
|
|
@ -960,7 +959,7 @@ def _import_delegations(self, parent_role):
|
|||
logger.debug('Adding roles delegated from ' + repr(parent_role) + '.')
|
||||
|
||||
# Iterate the keys of the delegated roles of 'parent_role' and load them.
|
||||
for keyid, keyinfo in six.iteritems(keys_info):
|
||||
for keyid, keyinfo in keys_info.items():
|
||||
if keyinfo['keytype'] in ['rsa', 'ed25519', 'ecdsa-sha2-nistp256']:
|
||||
|
||||
# We specify the keyid to ensure that it's the correct keyid
|
||||
|
|
@ -1205,7 +1204,7 @@ def _check_hashes(self, file_object, trusted_hashes):
|
|||
"""
|
||||
|
||||
# Verify each hash, raise an exception if any hash fails to verify
|
||||
for algorithm, trusted_hash in six.iteritems(trusted_hashes):
|
||||
for algorithm, trusted_hash in trusted_hashes.items():
|
||||
digest_object = sslib_hash.digest_fileobject(file_object,
|
||||
algorithm)
|
||||
computed_hash = digest_object.hexdigest()
|
||||
|
|
@ -1565,9 +1564,8 @@ def _get_metadata_file(self, metadata_role, remote_filename,
|
|||
". The update will continue as the major versions match.")
|
||||
|
||||
except (ValueError, TypeError) as error:
|
||||
six.raise_from(sslib_exceptions.FormatError('Improperly'
|
||||
' formatted spec_version, which must be in major.minor.fix format'),
|
||||
error)
|
||||
raise sslib_exceptions.FormatError('Improperly'
|
||||
' formatted spec_version, which must be in major.minor.fix format') from error
|
||||
|
||||
# If the version number is unspecified, ensure that the version number
|
||||
# downloaded is greater than the currently trusted version number for
|
||||
|
|
@ -2099,7 +2097,7 @@ def _fileinfo_has_changed(self, metadata_filename, new_fileinfo):
|
|||
# without having that result in considering all files as needing to be
|
||||
# updated, or not all hash algorithms listed can be calculated on the
|
||||
# specific client.
|
||||
for algorithm, hash_value in six.iteritems(new_fileinfo['hashes']):
|
||||
for algorithm, hash_value in new_fileinfo['hashes'].items():
|
||||
# We're only looking for a single match. This isn't a security
|
||||
# check, we just want to prevent unnecessary downloads.
|
||||
if algorithm in current_fileinfo['hashes']:
|
||||
|
|
@ -2403,7 +2401,7 @@ def _refresh_targets_metadata(self, rolename='targets',
|
|||
|
||||
if refresh_all_delegated_roles:
|
||||
|
||||
for role in six.iterkeys(self.metadata['current']['snapshot']['meta']):
|
||||
for role in self.metadata['current']['snapshot']['meta'].keys():
|
||||
# snapshot.json keeps track of root.json, targets.json, and delegated
|
||||
# roles (e.g., django.json, unclaimed.json). Remove the 'targets' role
|
||||
# because it gets updated when the targets.json file is updated in
|
||||
|
|
@ -2493,7 +2491,7 @@ def _targets_of_role(self, rolename, targets=None, skip_refresh=False):
|
|||
return []
|
||||
|
||||
# Get the targets specified by the role itself.
|
||||
for filepath, fileinfo in six.iteritems(self.metadata['current'][rolename].get('targets', [])):
|
||||
for filepath, fileinfo in self.metadata['current'][rolename].get('targets', []).items():
|
||||
new_target = {}
|
||||
new_target['filepath'] = filepath
|
||||
new_target['fileinfo'] = fileinfo
|
||||
|
|
@ -3084,7 +3082,7 @@ def updated_targets(self, targets, destination_directory):
|
|||
|
||||
# Try one of the algorithm/digest combos for a mismatch. We break
|
||||
# as soon as we find a mismatch.
|
||||
for algorithm, digest in six.iteritems(target['fileinfo']['hashes']):
|
||||
for algorithm, digest in target['fileinfo']['hashes'].items():
|
||||
digest_object = None
|
||||
try:
|
||||
digest_object = sslib_hash.digest_filename(target_filepath,
|
||||
|
|
|
|||
|
|
@ -37,7 +37,6 @@
|
|||
import shutil
|
||||
import tempfile
|
||||
import json
|
||||
import six
|
||||
|
||||
|
||||
import securesystemslib # pylint: disable=unused-import
|
||||
|
|
@ -948,7 +947,7 @@ def load_project(project_directory, prefix='', new_targets_location=None,
|
|||
roleinfo['expires'] = metadata_object['expires']
|
||||
roleinfo['paths'] = {}
|
||||
|
||||
for filepath, fileinfo in six.iteritems(metadata_object['targets']):
|
||||
for filepath, fileinfo in metadata_object['targets'].items():
|
||||
roleinfo['paths'].update({filepath: fileinfo.get('custom', {})})
|
||||
roleinfo['delegations'] = metadata_object['delegations']
|
||||
roleinfo['partial_loaded'] = False
|
||||
|
|
|
|||
|
|
@ -32,9 +32,9 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
import logging
|
||||
import six
|
||||
import timeit
|
||||
import tempfile
|
||||
from urllib import parse
|
||||
|
||||
import securesystemslib # pylint: disable=unused-import
|
||||
from securesystemslib import formats as sslib_formats
|
||||
|
|
@ -185,7 +185,7 @@ def _download_file(url, required_length, fetcher, STRICT_REQUIRED_LENGTH=True):
|
|||
# This converts it to the common format. unquote() replaces %xx escapes in a
|
||||
# url with their single-character equivalent. A back-slash may be encoded as
|
||||
# %5c in the url, which should also be replaced with a forward slash.
|
||||
url = six.moves.urllib.parse.unquote(url).replace('\\', '/')
|
||||
url = parse.unquote(url).replace('\\', '/')
|
||||
logger.info('Downloading: ' + repr(url))
|
||||
|
||||
# This is the temporary file that we will return to contain the contents of
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@
|
|||
from __future__ import division
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import six
|
||||
from urllib import parse
|
||||
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -296,10 +296,10 @@ def __init__(self, mirror_errors):
|
|||
def __str__(self):
|
||||
all_errors = 'No working mirror was found:'
|
||||
|
||||
for mirror_url, mirror_error in six.iteritems(self.mirror_errors):
|
||||
for mirror_url, mirror_error in self.mirror_errors.items():
|
||||
try:
|
||||
# http://docs.python.org/2/library/urlparse.html#urlparse.urlparse
|
||||
mirror_url_tokens = six.moves.urllib.parse.urlparse(mirror_url)
|
||||
mirror_url_tokens = parse.urlparse(mirror_url)
|
||||
|
||||
except Exception:
|
||||
logger.exception('Failed to parse mirror URL: ' + repr(mirror_url))
|
||||
|
|
|
|||
|
|
@ -76,8 +76,6 @@
|
|||
import tuf
|
||||
from tuf import exceptions
|
||||
|
||||
import six
|
||||
|
||||
# As per TUF spec 1.0.0 the spec version field must follow the Semantic
|
||||
# Versioning 2.0.0 (semver) format. The regex pattern is provided by semver.
|
||||
# https://semver.org/spec/v2.0.0.html#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string
|
||||
|
|
@ -638,9 +636,8 @@ def expiry_string_to_datetime(expires):
|
|||
try:
|
||||
return datetime.datetime.strptime(expires, "%Y-%m-%dT%H:%M:%SZ")
|
||||
except ValueError as error:
|
||||
six.raise_from(sslib_exceptions.FormatError(
|
||||
'Failed to parse ' + repr(expires) + ' as an expiry time'),
|
||||
error)
|
||||
raise sslib_exceptions.FormatError(
|
||||
'Failed to parse ' + repr(expires) + ' as an expiry time') from error
|
||||
|
||||
|
||||
|
||||
|
|
@ -781,7 +778,7 @@ def parse_base64(base64_string):
|
|||
'base64_string'.
|
||||
"""
|
||||
|
||||
if not isinstance(base64_string, six.string_types):
|
||||
if not isinstance(base64_string, str):
|
||||
message = 'Invalid argument: '+repr(base64_string)
|
||||
raise sslib_exceptions.FormatError(message)
|
||||
|
||||
|
|
@ -991,15 +988,14 @@ def check_signable_object_format(signable):
|
|||
role_type = signable['signed']['_type']
|
||||
|
||||
except (KeyError, TypeError) as error:
|
||||
six.raise_from(sslib_exceptions.FormatError(
|
||||
'Untyped signable object.'), error)
|
||||
raise sslib_exceptions.FormatError('Untyped signable object.') from error
|
||||
|
||||
try:
|
||||
schema = SCHEMAS_BY_TYPE[role_type]
|
||||
|
||||
except KeyError as error:
|
||||
six.raise_from(sslib_exceptions.FormatError(
|
||||
'Unrecognized type ' + repr(role_type)), error)
|
||||
raise sslib_exceptions.FormatError('Unrecognized type '
|
||||
+ repr(role_type)) from error
|
||||
|
||||
if not signable['signatures']:
|
||||
raise exceptions.UnsignedMetadataError('Signable object of type ' +
|
||||
|
|
|
|||
|
|
@ -52,8 +52,6 @@
|
|||
from tuf import exceptions
|
||||
from tuf import formats
|
||||
|
||||
import six
|
||||
|
||||
# List of strings representing the key types supported by TUF.
|
||||
_SUPPORTED_KEY_TYPES = ['rsa', 'ed25519', 'ecdsa-sha2-nistp256']
|
||||
|
||||
|
|
@ -119,7 +117,7 @@ def create_keydb_from_root_metadata(root_metadata, repository_name='default'):
|
|||
# Iterate the keys found in 'root_metadata' by converting them to
|
||||
# 'RSAKEY_SCHEMA' if their type is 'rsa', and then adding them to the
|
||||
# key database using the provided keyid.
|
||||
for keyid, key_metadata in six.iteritems(root_metadata['keys']):
|
||||
for keyid, key_metadata in root_metadata['keys'].items():
|
||||
if key_metadata['keytype'] in _SUPPORTED_KEY_TYPES:
|
||||
# 'key_metadata' is stored in 'KEY_SCHEMA' format. Call
|
||||
# create_from_metadata_format() to get the key in 'RSAKEY_SCHEMA' format,
|
||||
|
|
@ -348,7 +346,7 @@ def get_key(keyid, repository_name='default'):
|
|||
return copy.deepcopy(_keydb_dict[repository_name][keyid])
|
||||
|
||||
except KeyError as error:
|
||||
six.raise_from(exceptions.UnknownKeyError('Key: ' + keyid), error)
|
||||
raise exceptions.UnknownKeyError('Key: ' + keyid) from error
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
import os
|
||||
from urllib import parse
|
||||
|
||||
import securesystemslib # pylint: disable=unused-import
|
||||
from securesystemslib import exceptions as sslib_exceptions
|
||||
|
|
@ -39,7 +40,6 @@
|
|||
|
||||
from tuf import formats
|
||||
|
||||
import six
|
||||
|
||||
# The type of file to be downloaded from a repository. The
|
||||
# 'get_list_of_mirrors' function supports these file types.
|
||||
|
|
@ -98,7 +98,7 @@ def get_list_of_mirrors(file_type, file_path, mirrors_dict):
|
|||
path_key = 'metadata_path' if file_type == 'meta' else 'targets_path'
|
||||
|
||||
list_of_mirrors = []
|
||||
for junk, mirror_info in six.iteritems(mirrors_dict):
|
||||
for junk, mirror_info in mirrors_dict.items():
|
||||
# Does mirror serve this file type at all?
|
||||
path = mirror_info.get(path_key)
|
||||
if path is None:
|
||||
|
|
@ -114,12 +114,12 @@ def get_list_of_mirrors(file_type, file_path, mirrors_dict):
|
|||
confined_target_dirs):
|
||||
continue
|
||||
|
||||
# urllib.quote(string) replaces special characters in string using the %xx
|
||||
# parse.quote(string) replaces special characters in string using the %xx
|
||||
# escape. This is done to avoid parsing issues of the URL on the server
|
||||
# side. Do *NOT* pass URLs with Unicode characters without first encoding
|
||||
# the URL as UTF-8. We need a long-term solution with #61.
|
||||
# http://bugs.python.org/issue1712522
|
||||
file_path = six.moves.urllib.parse.quote(file_path)
|
||||
file_path = parse.quote(file_path)
|
||||
url = os.path.join(mirror_info['url_prefix'], path, file_path)
|
||||
|
||||
# The above os.path.join() result as well as input file_path may be
|
||||
|
|
|
|||
|
|
@ -37,7 +37,6 @@
|
|||
import logging
|
||||
import shutil
|
||||
import json
|
||||
import six
|
||||
import tempfile
|
||||
|
||||
import securesystemslib # pylint: disable=unused-import
|
||||
|
|
@ -542,8 +541,8 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
consistent_snapshot = root_metadata['consistent_snapshot']
|
||||
|
||||
except sslib_exceptions.StorageError as error:
|
||||
six.raise_from(exceptions.RepositoryError('Cannot load the required'
|
||||
' root file: ' + repr(root_filename)), error)
|
||||
raise exceptions.RepositoryError('Cannot load the required'
|
||||
' root file: ' + repr(root_filename)) from error
|
||||
|
||||
# Load 'timestamp.json'. A Timestamp role file without a version number is
|
||||
# always written.
|
||||
|
|
@ -571,8 +570,8 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
repository_name=repository_name)
|
||||
|
||||
except sslib_exceptions.StorageError as error:
|
||||
six.raise_from(exceptions.RepositoryError('Cannot load the Timestamp '
|
||||
'file: ' + repr(timestamp_filename)), error)
|
||||
raise exceptions.RepositoryError('Cannot load the Timestamp '
|
||||
'file: ' + repr(timestamp_filename)) from error
|
||||
|
||||
# Load 'snapshot.json'. A consistent snapshot.json must be calculated if
|
||||
# 'consistent_snapshot' is True.
|
||||
|
|
@ -617,8 +616,8 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
repository_name=repository_name)
|
||||
|
||||
except sslib_exceptions.StorageError as error:
|
||||
six.raise_from(exceptions.RepositoryError('The Snapshot file '
|
||||
'cannot be loaded: '+ repr(snapshot_filename)), error)
|
||||
raise exceptions.RepositoryError('The Snapshot file '
|
||||
'cannot be loaded: '+ repr(snapshot_filename)) from error
|
||||
|
||||
# Load 'targets.json'. A consistent snapshot of the Targets role must be
|
||||
# calculated if 'consistent_snapshot' is True.
|
||||
|
|
@ -661,7 +660,7 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
repository_name=repository_name)
|
||||
|
||||
# Add the keys specified in the delegations field of the Targets role.
|
||||
for keyid, key_metadata in six.iteritems(targets_metadata['delegations']['keys']):
|
||||
for keyid, key_metadata in targets_metadata['delegations']['keys'].items():
|
||||
|
||||
# Use the keyid found in the delegation
|
||||
key_object, _ = sslib_keys.format_metadata_to_key(key_metadata,
|
||||
|
|
@ -680,8 +679,8 @@ def _load_top_level_metadata(repository, top_level_filenames, repository_name):
|
|||
pass
|
||||
|
||||
except sslib_exceptions.StorageError as error:
|
||||
six.raise_from(exceptions.RepositoryError('The Targets file '
|
||||
'can not be loaded: ' + repr(targets_filename)), error)
|
||||
raise exceptions.RepositoryError('The Targets file '
|
||||
'can not be loaded: ' + repr(targets_filename)) from error
|
||||
|
||||
return repository, consistent_snapshot
|
||||
|
||||
|
|
@ -1428,7 +1427,7 @@ def generate_targets_metadata(targets_directory, target_files, version,
|
|||
if use_existing_fileinfo:
|
||||
# Use the provided fileinfo dicts, conforming to FILEINFO_SCHEMA, rather than
|
||||
# generating fileinfo
|
||||
for target, fileinfo in six.iteritems(target_files):
|
||||
for target, fileinfo in target_files.items():
|
||||
|
||||
# Ensure all fileinfo entries in target_files have a non-empty hashes dict
|
||||
if not fileinfo.get('hashes', None):
|
||||
|
|
@ -1497,7 +1496,7 @@ def _generate_targets_fileinfo(target_files, targets_directory,
|
|||
filedict = {}
|
||||
|
||||
# Generate the fileinfo of all the target files listed in 'target_files'.
|
||||
for target, fileinfo in six.iteritems(target_files):
|
||||
for target, fileinfo in target_files.items():
|
||||
|
||||
# The root-most folder of the targets directory should not be included in
|
||||
# target paths listed in targets metadata.
|
||||
|
|
@ -1518,7 +1517,7 @@ def _generate_targets_fileinfo(target_files, targets_directory,
|
|||
|
||||
# Copy 'target_path' to 'digest_target' if consistent hashing is enabled.
|
||||
if write_consistent_targets:
|
||||
for target_digest in six.itervalues(filedict[relative_targetpath]['hashes']):
|
||||
for target_digest in filedict[relative_targetpath]['hashes'].values():
|
||||
dirname, basename = os.path.split(target_path)
|
||||
digest_filename = target_digest + '.' + basename
|
||||
digest_target = os.path.join(dirname, digest_filename)
|
||||
|
|
|
|||
|
|
@ -39,7 +39,6 @@
|
|||
import tempfile
|
||||
import shutil
|
||||
import json
|
||||
import six
|
||||
|
||||
from collections import deque
|
||||
|
||||
|
|
@ -3160,7 +3159,7 @@ def load_repository(repository_directory, repository_name='default',
|
|||
# log a warning here as there may be many such duplicate key warnings.
|
||||
# The repository maintainer should have also been made aware of the
|
||||
# duplicate key when it was added.
|
||||
for key_metadata in six.itervalues(metadata_object['delegations']['keys']):
|
||||
for key_metadata in metadata_object['delegations']['keys'].values():
|
||||
|
||||
# The repo may have used hashing algorithms for the generated keyids
|
||||
# that doesn't match the client's set of hash algorithms. Make sure
|
||||
|
|
|
|||
|
|
@ -7,9 +7,9 @@
|
|||
|
||||
# Imports
|
||||
import requests
|
||||
import six
|
||||
import logging
|
||||
import time
|
||||
from urllib import parse
|
||||
from urllib3.exceptions import ReadTimeoutError
|
||||
|
||||
import tuf
|
||||
|
|
@ -137,7 +137,7 @@ def _get_session(self, url):
|
|||
"""
|
||||
# Use a different requests.Session per schema+hostname combination, to
|
||||
# reuse connections while minimizing subtle security issues.
|
||||
parsed_url = six.moves.urllib.parse.urlparse(url)
|
||||
parsed_url = parse.urlparse(url)
|
||||
|
||||
if not parsed_url.scheme or not parsed_url.hostname:
|
||||
raise exceptions.URLParsingError(
|
||||
|
|
|
|||
|
|
@ -59,8 +59,6 @@
|
|||
from tuf import exceptions
|
||||
from tuf import formats
|
||||
|
||||
import six
|
||||
|
||||
# See 'tuf.log' to learn how logging is handled in TUF.
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -135,7 +133,7 @@ def create_roledb_from_root_metadata(root_metadata, repository_name='default'):
|
|||
|
||||
# Iterate the roles found in 'root_metadata' and add them to '_roledb_dict'.
|
||||
# Duplicates are avoided.
|
||||
for rolename, roleinfo in six.iteritems(root_metadata['roles']):
|
||||
for rolename, roleinfo in root_metadata['roles'].items():
|
||||
if rolename == 'root':
|
||||
roleinfo['version'] = root_metadata['version']
|
||||
roleinfo['expires'] = root_metadata['expires']
|
||||
|
|
|
|||
|
|
@ -146,7 +146,6 @@
|
|||
import shutil
|
||||
import time
|
||||
import fnmatch
|
||||
import six
|
||||
|
||||
import securesystemslib # pylint: disable=unused-import
|
||||
from securesystemslib import exceptions as sslib_exceptions
|
||||
|
|
@ -469,9 +468,9 @@ def import_privatekey_from_file(keypath, password=None):
|
|||
encrypted_key, 'rsassa-pss-sha256', password)
|
||||
|
||||
except sslib_exceptions.CryptoError as error:
|
||||
six.raise_from(exceptions.Error(repr(keypath) + ' cannot be '
|
||||
raise exceptions.Error(repr(keypath) + ' cannot be '
|
||||
' imported, possibly because an invalid key file is given or '
|
||||
' the decryption password is incorrect.'), error)
|
||||
' the decryption password is incorrect.') from error
|
||||
|
||||
if key_object['keytype'] not in SUPPORTED_KEY_TYPES:
|
||||
raise exceptions.Error('Trying to import an unsupported key'
|
||||
|
|
@ -752,7 +751,7 @@ def remove_target_files_from_metadata(parsed_arguments, repository):
|
|||
parsed_arguments.role, repository._repository_name)
|
||||
|
||||
for glob_pattern in parsed_arguments.remove:
|
||||
for path in list(six.iterkeys(roleinfo['paths'])):
|
||||
for path in list(roleinfo['paths'].keys()):
|
||||
if fnmatch.fnmatch(path, glob_pattern):
|
||||
del roleinfo['paths'][path]
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue