mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
Now crypto related files are in ssl_crypto and ssl_commons
This commit is contained in:
Artiom Baloian
parent
1bf7ec1135
commit
299ff68edd
26 changed files with 313 additions and 2242 deletions
|
|
@ -54,7 +54,7 @@
|
|||
import tuf.tufformats
|
||||
import tuf.ssl_crypto.util
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.log
|
||||
import tuf.client.updater as updater
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
|
|
@ -166,7 +166,7 @@ def tearDown(self):
|
|||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
# updater.Updater() populates the roledb with the name "test_repository"
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
def test_without_tuf(self):
|
||||
# Verify that a target file replaced with a malicious version is downloaded
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@
|
|||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.developer_tool as developer_tool
|
||||
|
||||
from tuf.developer_tool import METADATA_DIRECTORY_NAME
|
||||
|
|
@ -56,7 +56,7 @@ def setUp(self):
|
|||
def tearDown(self):
|
||||
# called after every test case
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_create_new_project(self):
|
||||
|
|
@ -152,7 +152,7 @@ def test_create_new_project(self):
|
|||
os.chmod(local_tmp, 0o0555)
|
||||
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.assertRaises(OSError, developer_tool.create_new_project ,project_name,
|
||||
metadata_directory, location_in_repository, targets_directory,
|
||||
project_key)
|
||||
|
|
@ -163,7 +163,7 @@ def test_create_new_project(self):
|
|||
os.chmod(local_tmp, 0o0555)
|
||||
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.assertRaises(OSError, developer_tool.create_new_project ,project_name,
|
||||
metadata_directory, location_in_repository, targets_directory,
|
||||
project_key)
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@
|
|||
import tuf.client.updater as updater
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
from simple_settings import settings
|
||||
import six
|
||||
|
||||
|
|
@ -167,7 +167,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_without_tuf(self):
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@
|
|||
import tuf.log
|
||||
import tuf.client.updater as updater
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
from simple_settings import settings
|
||||
import six
|
||||
|
|
@ -172,7 +172,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@
|
|||
import tuf.repository_tool as repo_tool
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
from simple_settings import settings
|
||||
|
||||
import six
|
||||
|
|
@ -181,7 +181,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_without_tuf(self):
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@
|
|||
import tuf.interposition.configuration as configuration
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
|
||||
|
||||
if sys.version_info >= (2, 7):
|
||||
|
|
@ -196,7 +196,7 @@ def tearDown(self):
|
|||
# We are inheriting from custom class.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
# Unit Tests
|
||||
|
|
@ -417,7 +417,7 @@ def tearDown(self):
|
|||
# We are inheriting from custom class.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb('localhost')
|
||||
tuf.keydb.clear_keydb('localhost')
|
||||
tuf.ssl_crypto.keydb.clear_keydb('localhost')
|
||||
|
||||
|
||||
# Unit Tests
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@
|
|||
import tuf.log
|
||||
from simple_settings import settings
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.repository_tool as repo_tool
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.client.updater as updater
|
||||
|
|
@ -179,7 +179,7 @@ def tearDown(self):
|
|||
# We are inheriting from custom class.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -30,8 +30,8 @@
|
|||
|
||||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.keys
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.log
|
||||
from simple_settings import settings
|
||||
|
||||
|
|
@ -41,7 +41,7 @@
|
|||
# Generate the three keys to use in our test cases.
|
||||
KEYS = []
|
||||
for junk in range(3):
|
||||
rsa_key = tuf.keys.generate_rsa_key(2048)
|
||||
rsa_key = tuf.ssl_crypto.keys.generate_rsa_key(2048)
|
||||
rsa_key['keyid_hash_algorithms'] = settings.REPOSITORY_HASH_ALGORITHMS
|
||||
KEYS.append(rsa_key)
|
||||
|
||||
|
|
@ -49,12 +49,12 @@
|
|||
|
||||
class TestKeydb(unittest.TestCase):
|
||||
def setUp(self):
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
def tearDown(self):
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
@ -63,19 +63,19 @@ def test_create_keydb(self):
|
|||
repository_name = 'example_repository'
|
||||
|
||||
# The keydb dictionary should contain only the 'default' repository entry.
|
||||
self.assertTrue('default' in tuf.keydb._keydb_dict)
|
||||
self.assertEqual(1, len(tuf.keydb._keydb_dict))
|
||||
self.assertTrue('default' in tuf.ssl_crypto.keydb._keydb_dict)
|
||||
self.assertEqual(1, len(tuf.ssl_crypto.keydb._keydb_dict))
|
||||
|
||||
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
self.assertEqual(2, len(tuf.keydb._keydb_dict))
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
self.assertEqual(2, len(tuf.ssl_crypto.keydb._keydb_dict))
|
||||
|
||||
# Verify that a keydb cannot be created for a name that already exists.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.create_keydb, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.create_keydb, repository_name)
|
||||
|
||||
# Ensure that the key database for 'example_repository' is deleted so that
|
||||
# the key database is returned to its original, default state.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
|
||||
|
||||
|
|
@ -85,41 +85,41 @@ def test_remove_keydb(self):
|
|||
keyid = KEYS[0]['keyid']
|
||||
|
||||
repository_name = 'example_repository'
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.remove_keydb, 'default')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.remove_keydb, 'default')
|
||||
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
# tuf.keydb.remove_keydb() logs a warning if a keydb for a non-existent
|
||||
# tuf.ssl_crypto.keydb.remove_keydb() logs a warning if a keydb for a non-existent
|
||||
# repository is specified.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
# Test condition for improperly formatted argument, and unexpected argument.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_keydb, 123)
|
||||
self.assertRaises(TypeError, tuf.keydb.remove_keydb, rsakey, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_keydb, 123)
|
||||
self.assertRaises(TypeError, tuf.ssl_crypto.keydb.remove_keydb, rsakey, 123)
|
||||
|
||||
|
||||
|
||||
def test_clear_keydb(self):
|
||||
# Test condition ensuring 'clear_keydb()' clears the keydb database.
|
||||
# Test the length of the keydb before and after adding a key.
|
||||
self.assertEqual(0, len(tuf.keydb._keydb_dict['default']))
|
||||
self.assertEqual(0, len(tuf.ssl_crypto.keydb._keydb_dict['default']))
|
||||
rsakey = KEYS[0]
|
||||
keyid = KEYS[0]['keyid']
|
||||
tuf.keydb._keydb_dict['default'][keyid] = rsakey
|
||||
self.assertEqual(1, len(tuf.keydb._keydb_dict['default']))
|
||||
tuf.keydb.clear_keydb()
|
||||
self.assertEqual(0, len(tuf.keydb._keydb_dict['default']))
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][keyid] = rsakey
|
||||
self.assertEqual(1, len(tuf.ssl_crypto.keydb._keydb_dict['default']))
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.assertEqual(0, len(tuf.ssl_crypto.keydb._keydb_dict['default']))
|
||||
|
||||
# Test condition for unexpected argument.
|
||||
self.assertRaises(TypeError, tuf.keydb.clear_keydb, 'default', False, 'unexpected_argument')
|
||||
self.assertRaises(TypeError, tuf.ssl_crypto.keydb.clear_keydb, 'default', False, 'unexpected_argument')
|
||||
|
||||
# Test condition for improperly formatted arguments.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.clear_keydb, 0)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.clear_keydb, 'default', 0)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.clear_keydb, 0)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.clear_keydb, 'default', 0)
|
||||
|
||||
# Test condition for non-existent repository name.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.clear_keydb, 'non-existent')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.clear_keydb, 'non-existent')
|
||||
|
||||
# Test condition for keys added to a non-default key database. Unlike the
|
||||
# test conditions above, this test makes use of the public functions
|
||||
|
|
@ -128,17 +128,17 @@ def test_clear_keydb(self):
|
|||
rsakey = KEYS[0]
|
||||
keyid = KEYS[0]['keyid']
|
||||
repository_name = 'example_repository'
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid, repository_name)
|
||||
tuf.keydb.add_key(rsakey, keyid, repository_name)
|
||||
self.assertEqual(rsakey, tuf.keydb.get_key(keyid, repository_name))
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid, repository_name)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey, keyid, repository_name)
|
||||
self.assertEqual(rsakey, tuf.ssl_crypto.keydb.get_key(keyid, repository_name))
|
||||
|
||||
tuf.keydb.clear_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid, repository_name)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid, repository_name)
|
||||
|
||||
# Remove 'repository_name' from the key database to revert it back to its
|
||||
# original, default state (i.e., only the 'default' repository exists).
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
|
||||
|
||||
|
|
@ -146,46 +146,46 @@ def test_get_key(self):
|
|||
# Test conditions using valid 'keyid' arguments.
|
||||
rsakey = KEYS[0]
|
||||
keyid = KEYS[0]['keyid']
|
||||
tuf.keydb._keydb_dict['default'][keyid] = rsakey
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][keyid] = rsakey
|
||||
rsakey2 = KEYS[1]
|
||||
keyid2 = KEYS[1]['keyid']
|
||||
tuf.keydb._keydb_dict['default'][keyid2] = rsakey2
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][keyid2] = rsakey2
|
||||
|
||||
self.assertEqual(rsakey, tuf.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.keydb.get_key(keyid2))
|
||||
self.assertNotEqual(rsakey2, tuf.keydb.get_key(keyid))
|
||||
self.assertNotEqual(rsakey, tuf.keydb.get_key(keyid2))
|
||||
self.assertEqual(rsakey, tuf.ssl_crypto.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.ssl_crypto.keydb.get_key(keyid2))
|
||||
self.assertNotEqual(rsakey2, tuf.ssl_crypto.keydb.get_key(keyid))
|
||||
self.assertNotEqual(rsakey, tuf.ssl_crypto.keydb.get_key(keyid2))
|
||||
|
||||
# Test conditions using invalid arguments.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, None)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, ['123'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, {'keyid': '123'})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, '')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.get_key, keyid, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, None)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, ['123'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, {'keyid': '123'})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, '')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.get_key, keyid, 123)
|
||||
|
||||
# Test condition using a 'keyid' that has not been added yet.
|
||||
keyid3 = KEYS[2]['keyid']
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid3)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid3)
|
||||
|
||||
# Test condition for a key added to a non-default repository.
|
||||
repository_name = 'example_repository'
|
||||
rsakey3 = KEYS[2]
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
tuf.keydb.add_key(rsakey3, keyid3, repository_name)
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey3, keyid3, repository_name)
|
||||
|
||||
# Test condition for a key added to a non-existent repository.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.get_key,
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.get_key,
|
||||
keyid, 'non-existent')
|
||||
|
||||
# Verify that 'rsakey3' is added to the expected repository name.
|
||||
# If not supplied, the 'default' repository name is searched.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid3)
|
||||
self.assertEqual(rsakey3, tuf.keydb.get_key(keyid3, repository_name))
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid3)
|
||||
self.assertEqual(rsakey3, tuf.ssl_crypto.keydb.get_key(keyid3, repository_name))
|
||||
|
||||
# Remove the 'example_repository' so that other test functions have access
|
||||
# to a default state of the keydb.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
|
||||
|
||||
|
|
@ -197,55 +197,55 @@ def test_add_key(self):
|
|||
keyid2 = KEYS[1]['keyid']
|
||||
rsakey3 = KEYS[2]
|
||||
keyid3 = KEYS[2]['keyid']
|
||||
self.assertEqual(None, tuf.keydb.add_key(rsakey, keyid))
|
||||
self.assertEqual(None, tuf.keydb.add_key(rsakey2, keyid2))
|
||||
self.assertEqual(None, tuf.keydb.add_key(rsakey3))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.add_key(rsakey, keyid))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.add_key(rsakey2, keyid2))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.add_key(rsakey3))
|
||||
|
||||
self.assertEqual(rsakey, tuf.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.keydb.get_key(keyid2))
|
||||
self.assertEqual(rsakey3, tuf.keydb.get_key(keyid3))
|
||||
self.assertEqual(rsakey, tuf.ssl_crypto.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.ssl_crypto.keydb.get_key(keyid2))
|
||||
self.assertEqual(rsakey3, tuf.ssl_crypto.keydb.get_key(keyid3))
|
||||
|
||||
# Test conditions using arguments with invalid formats.
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
rsakey3['keytype'] = 'bad_keytype'
|
||||
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, None, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, '', keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, ['123'], keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, {'a': 'b'}, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey, {'keyid': ''})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey, False)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey, ['keyid'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey3, keyid3)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, None, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, '', keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, ['123'], keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, {'a': 'b'}, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey, {'keyid': ''})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey, False)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey, ['keyid'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey3, keyid3)
|
||||
rsakey3['keytype'] = 'rsa'
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.add_key, rsakey3, keyid3, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.add_key, rsakey3, keyid3, 123)
|
||||
|
||||
# Test conditions where keyid does not match the rsakey.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.keydb.add_key, rsakey, keyid2)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.keydb.add_key, rsakey2, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.ssl_crypto.keydb.add_key, rsakey, keyid2)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.ssl_crypto.keydb.add_key, rsakey2, keyid)
|
||||
|
||||
# Test conditions using keyids that have already been added.
|
||||
tuf.keydb.add_key(rsakey, keyid)
|
||||
tuf.keydb.add_key(rsakey2, keyid2)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.KeyAlreadyExistsError, tuf.keydb.add_key, rsakey)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.KeyAlreadyExistsError, tuf.keydb.add_key, rsakey2)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey, keyid)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey2, keyid2)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.KeyAlreadyExistsError, tuf.ssl_crypto.keydb.add_key, rsakey)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.KeyAlreadyExistsError, tuf.ssl_crypto.keydb.add_key, rsakey2)
|
||||
|
||||
# Test condition for key added to the keydb of a non-default repository.
|
||||
repository_name = 'example_repository'
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid3, repository_name)
|
||||
tuf.keydb.add_key(rsakey3, keyid3, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid3)
|
||||
self.assertEqual(rsakey3, tuf.keydb.get_key(keyid3, repository_name))
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid3, repository_name)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey3, keyid3, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid3)
|
||||
self.assertEqual(rsakey3, tuf.ssl_crypto.keydb.get_key(keyid3, repository_name))
|
||||
|
||||
# Test condition for key added to the keydb of a non-existent repository.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.add_key,
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.add_key,
|
||||
rsakey3, keyid3, 'non-existent')
|
||||
|
||||
# Reset the keydb to its original, default state. Other test functions
|
||||
# expect only the 'default' repository to exist.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
|
||||
|
||||
|
|
@ -257,43 +257,43 @@ def test_remove_key(self):
|
|||
keyid2 = KEYS[1]['keyid']
|
||||
rsakey3 = KEYS[2]
|
||||
keyid3 = KEYS[2]['keyid']
|
||||
tuf.keydb.add_key(rsakey, keyid)
|
||||
tuf.keydb.add_key(rsakey2, keyid2)
|
||||
tuf.keydb.add_key(rsakey3, keyid3)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey, keyid)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey2, keyid2)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey3, keyid3)
|
||||
|
||||
self.assertEqual(None, tuf.keydb.remove_key(keyid))
|
||||
self.assertEqual(None, tuf.keydb.remove_key(keyid2))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.remove_key(keyid))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.remove_key(keyid2))
|
||||
|
||||
# Ensure the keys were actually removed.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid2)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid2)
|
||||
|
||||
# Test for 'keyid' not in keydb.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.remove_key, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.remove_key, keyid)
|
||||
|
||||
# Test condition for unknown key argument.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.remove_key, '1')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.remove_key, '1')
|
||||
|
||||
# Test condition for removal of keys from a non-default repository.
|
||||
repository_name = 'example_repository'
|
||||
tuf.keydb.create_keydb(repository_name)
|
||||
tuf.keydb.add_key(rsakey, keyid, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.keydb.remove_key, keyid, 'non-existent')
|
||||
tuf.keydb.remove_key(keyid, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.remove_key, keyid, repository_name)
|
||||
tuf.ssl_crypto.keydb.create_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.add_key(rsakey, keyid, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.InvalidNameError, tuf.ssl_crypto.keydb.remove_key, keyid, 'non-existent')
|
||||
tuf.ssl_crypto.keydb.remove_key(keyid, repository_name)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.remove_key, keyid, repository_name)
|
||||
|
||||
# Reset the keydb so that subsequent tests have access to the original,
|
||||
# default keydb.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
|
||||
# Test conditions for arguments with invalid formats.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, None)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, '')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, ['123'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, keyid, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.keydb.remove_key, {'bad': '123'})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.keydb.remove_key, rsakey3)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, None)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, '')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, ['123'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, keyid, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.ssl_crypto.keydb.remove_key, {'bad': '123'})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.ssl_crypto.keydb.remove_key, rsakey3)
|
||||
|
||||
|
||||
|
||||
|
|
@ -318,44 +318,44 @@ def test_create_keydb_from_root_metadata(self):
|
|||
keydict, roledict,
|
||||
consistent_snapshot,
|
||||
compression_algorithms)
|
||||
self.assertEqual(None, tuf.keydb.create_keydb_from_root_metadata(root_metadata))
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata))
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
|
||||
# Ensure 'keyid' and 'keyid2' were added to the keydb database.
|
||||
self.assertEqual(rsakey, tuf.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.keydb.get_key(keyid2))
|
||||
self.assertEqual(rsakey, tuf.ssl_crypto.keydb.get_key(keyid))
|
||||
self.assertEqual(rsakey2, tuf.ssl_crypto.keydb.get_key(keyid2))
|
||||
|
||||
# Verify that the keydb is populated for a non-default repository.
|
||||
repository_name = 'example_repository'
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_metadata, repository_name)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata, repository_name)
|
||||
|
||||
# Test conditions for arguments with invalid formats.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, None)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, None)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, '')
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, '')
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, 123)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, 123)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, ['123'])
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, ['123'])
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, {'bad': '123'})
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, {'bad': '123'})
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError,
|
||||
tuf.keydb.create_keydb_from_root_metadata, root_metadata, 123)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata, root_metadata, 123)
|
||||
|
||||
# Verify that a keydb cannot be created for a non-existent repository name.
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_metadata, 'non-existent')
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata, 'non-existent')
|
||||
|
||||
# Remove the 'non-existent' and 'example_repository' key database so that
|
||||
# subsequent test functions have access to a default keydb.
|
||||
tuf.keydb.remove_keydb(repository_name)
|
||||
tuf.keydb.remove_keydb('non-existent')
|
||||
tuf.ssl_crypto.keydb.remove_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.remove_keydb('non-existent')
|
||||
|
||||
|
||||
# Test conditions for correctly formatted 'root_metadata' arguments but
|
||||
# containing incorrect keyids or key types. In these conditions, the keys
|
||||
# should not be added to the keydb database and a warning should be logged.
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
# 'keyid' does not match 'rsakey2'.
|
||||
keydict[keyid] = rsakey2
|
||||
|
|
@ -374,13 +374,13 @@ def test_create_keydb_from_root_metadata(self):
|
|||
keydict, roledict,
|
||||
consistent_snapshot,
|
||||
compression_algorithms)
|
||||
self.assertEqual(None, tuf.keydb.create_keydb_from_root_metadata(root_metadata))
|
||||
self.assertEqual(None, tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata))
|
||||
|
||||
# Ensure only 'keyid2' was added to the keydb database. 'keyid' and
|
||||
# 'keyid3' should not be stored.
|
||||
self.assertEqual(rsakey2, tuf.keydb.get_key(keyid2))
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.keydb.get_key, keyid3)
|
||||
self.assertEqual(rsakey2, tuf.ssl_crypto.keydb.get_key(keyid2))
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid)
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.UnknownKeyError, tuf.ssl_crypto.keydb.get_key, keyid3)
|
||||
rsakey3['keytype'] = 'rsa'
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -33,11 +33,11 @@
|
|||
import tuf.ssl_crypto.pycrypto_keys
|
||||
import tuf.ssl_crypto.formats
|
||||
import tuf.tufformats
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
|
||||
logger = logging.getLogger('tuf.test_keys')
|
||||
|
||||
KEYS = tuf.keys
|
||||
KEYS = tuf.ssl_crypto.keys
|
||||
FORMAT_ERROR_MSG = 'tuf.ssl_commons.exceptions.FormatError was raised! Check object\'s format.'
|
||||
DATA = 'SOME DATA REQUIRING AUTHENTICITY.'
|
||||
|
||||
|
|
@ -262,7 +262,7 @@ def test_verify_signature(self):
|
|||
self.assertRaises(TypeError, KEYS.verify_signature)
|
||||
|
||||
# Verify that the pure python 'ed25519' base case (triggered if 'pynacl' is
|
||||
# unavailable) is executed in tuf.keys.verify_signature().
|
||||
# unavailable) is executed in tuf.ssl_crypto.keys.verify_signature().
|
||||
KEYS._ED25519_CRYPTO_LIBRARY = 'invalid'
|
||||
KEYS._available_crypto_libraries = ['invalid']
|
||||
verified = KEYS.verify_signature(self.ed25519key_dict, ed25519_signature, DATA)
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@
|
|||
import tuf.repository_tool as repo_tool
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
from simple_settings import settings
|
||||
|
||||
import six
|
||||
|
|
@ -175,7 +175,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_with_tuf(self):
|
||||
|
|
|
|||
|
|
@ -173,7 +173,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_without_tuf(self):
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@
|
|||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.ssl_crypto.hash
|
||||
from simple_settings import settings
|
||||
import tuf.repository_lib as repo_lib
|
||||
|
|
@ -91,7 +91,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
@ -273,7 +273,7 @@ def test_import_ed25519_publickey_from_file(self):
|
|||
keytype = imported_ed25519_key['keytype']
|
||||
keyval = imported_ed25519_key['keyval']
|
||||
ed25519key_metadata_format = \
|
||||
tuf.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
tuf.ssl_crypto.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
|
||||
ed25519key_metadata_format['keytype'] = 'invalid_keytype'
|
||||
with open(ed25519_keypath + '.pub', 'wb') as file_object:
|
||||
|
|
@ -428,9 +428,9 @@ def test_generate_root_metadata(self):
|
|||
root_signable = tuf.ssl_crypto.util.load_json_file(root_filepath)
|
||||
|
||||
# generate_root_metadata() expects the top-level roles and keys to be
|
||||
# available in 'tuf.keydb' and 'tuf.roledb'.
|
||||
# available in 'tuf.ssl_crypto.keydb' and 'tuf.roledb'.
|
||||
tuf.roledb.create_roledb_from_root_metadata(root_signable['signed'])
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_signable['signed'])
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_signable['signed'])
|
||||
expires = '1985-10-21T01:22:00Z'
|
||||
|
||||
root_metadata = repo_lib.generate_root_metadata(1, expires,
|
||||
|
|
@ -438,13 +438,13 @@ def test_generate_root_metadata(self):
|
|||
self.assertTrue(tuf.ssl_crypto.formats.ROOT_SCHEMA.matches(root_metadata))
|
||||
|
||||
root_keyids = tuf.roledb.get_role_keyids('root')
|
||||
tuf.keydb._keydb_dict['default'][root_keyids[0]]['keytype'] = 'bad_keytype'
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][root_keyids[0]]['keytype'] = 'bad_keytype'
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, repo_lib.generate_root_metadata, 1,
|
||||
expires, consistent_snapshot=False)
|
||||
|
||||
# Reset the root key's keytype, so that we can next verify that a different
|
||||
# tuf.ssl_commons.exceptions.Error exception is raised for duplicate keyids.
|
||||
tuf.keydb._keydb_dict['default'][root_keyids[0]]['keytype'] = 'rsa'
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][root_keyids[0]]['keytype'] = 'rsa'
|
||||
|
||||
# Add duplicate keyid to root's roleinfo.
|
||||
tuf.roledb._roledb_dict['default']['root']['keyids'].append(root_keyids[0])
|
||||
|
|
@ -461,7 +461,7 @@ def test_generate_root_metadata(self):
|
|||
|
||||
# Test for missing required roles and keys.
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, repo_lib.generate_root_metadata,
|
||||
1, expires, False)
|
||||
|
||||
|
|
@ -681,7 +681,7 @@ def test_sign_metadata(self):
|
|||
targets_filename = os.path.join(metadata_path, 'targets.json')
|
||||
targets_metadata = tuf.ssl_crypto.util.load_json_file(targets_filename)['signed']
|
||||
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
tuf.roledb.create_roledb_from_root_metadata(root_metadata)
|
||||
root_keyids = tuf.roledb.get_role_keyids('root')
|
||||
targets_keyids = tuf.roledb.get_role_keyids('targets')
|
||||
|
|
@ -696,13 +696,13 @@ def test_sign_metadata(self):
|
|||
repo_lib.import_ed25519_publickey_from_file(targets_public_keypath)
|
||||
|
||||
# sign_metadata() expects the private key 'root_metadata' to be in
|
||||
# 'tuf.keydb'. Remove any public keys that may be loaded before
|
||||
# 'tuf.ssl_crypto.keydb'. Remove any public keys that may be loaded before
|
||||
# adding private key, otherwise a 'tuf.KeyAlreadyExists' exception is
|
||||
# raised.
|
||||
tuf.keydb.remove_key(root_private_key['keyid'])
|
||||
tuf.keydb.add_key(root_private_key)
|
||||
tuf.keydb.remove_key(targets_public_key['keyid'])
|
||||
tuf.keydb.add_key(targets_public_key)
|
||||
tuf.ssl_crypto.keydb.remove_key(root_private_key['keyid'])
|
||||
tuf.ssl_crypto.keydb.add_key(root_private_key)
|
||||
tuf.ssl_crypto.keydb.remove_key(targets_public_key['keyid'])
|
||||
tuf.ssl_crypto.keydb.add_key(targets_public_key)
|
||||
|
||||
# Verify that a valid root signable is generated.
|
||||
root_signable = repo_lib.sign_metadata(root_metadata, root_keyids,
|
||||
|
|
@ -715,7 +715,7 @@ def test_sign_metadata(self):
|
|||
|
||||
# Add an invalid keytype to one of the root keys.
|
||||
root_keyid = root_keyids[0]
|
||||
tuf.keydb._keydb_dict['default'][root_keyid]['keytype'] = 'bad_keytype'
|
||||
tuf.ssl_crypto.keydb._keydb_dict['default'][root_keyid]['keytype'] = 'bad_keytype'
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.Error, repo_lib.sign_metadata, root_metadata,
|
||||
root_keyids, root_filename)
|
||||
|
||||
|
|
@ -999,7 +999,7 @@ def test__delete_obsolete_metadata(self):
|
|||
|
||||
def test__load_top_level_metadata(self):
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
temporary_directory = tempfile.mkdtemp(dir=self.temporary_directory)
|
||||
repository_directory = os.path.join(temporary_directory, 'repository')
|
||||
|
|
@ -1067,15 +1067,15 @@ def test__remove_invalid_and_duplicate_signatures(self):
|
|||
root_rsa_key = repo_lib.import_rsa_privatekey_from_file(key_filepath,
|
||||
'password')
|
||||
|
||||
# Add 'root_rsa_key' to tuf.keydb, since
|
||||
# Add 'root_rsa_key' to tuf.ssl_crypto.keydb, since
|
||||
# _remove_invalid_and_duplicate_signatures() checks for unknown keys in
|
||||
# tuf.keydb.
|
||||
tuf.keydb.add_key(root_rsa_key)
|
||||
# tuf.ssl_crypto.keydb.
|
||||
tuf.ssl_crypto.keydb.add_key(root_rsa_key)
|
||||
|
||||
# Append the new valid, but duplicate PSS signature, and test that
|
||||
# duplicates are removed. create_signature() generates a key for the
|
||||
# key type of the first argument (i.e., root_rsa_key).
|
||||
new_pss_signature = tuf.keys.create_signature(root_rsa_key,
|
||||
new_pss_signature = tuf.ssl_crypto.keys.create_signature(root_rsa_key,
|
||||
root_signable['signed'])
|
||||
root_signable['signatures'].append(new_pss_signature)
|
||||
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@
|
|||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.ssl_crypto.hash
|
||||
import tuf.repository_tool as repo_tool
|
||||
|
||||
|
|
@ -88,7 +88,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_init(self):
|
||||
|
|
@ -438,7 +438,7 @@ def __init__(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.metadata = None
|
||||
|
||||
|
||||
|
|
@ -785,7 +785,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
|
||||
|
||||
|
|
@ -807,7 +807,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
|
||||
|
||||
|
|
@ -832,7 +832,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
|
||||
|
||||
|
|
@ -887,7 +887,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
self.targets_object = None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@
|
|||
|
||||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.roledb
|
||||
import tuf.log
|
||||
|
||||
|
|
@ -40,7 +40,7 @@
|
|||
# Generate the three keys to use in our test cases.
|
||||
KEYS = []
|
||||
for junk in range(3):
|
||||
KEYS.append(tuf.keys.generate_rsa_key(2048))
|
||||
KEYS.append(tuf.ssl_crypto.keys.generate_rsa_key(2048))
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@
|
|||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.ssl_crypto.hash
|
||||
import tuf.repository_tool as repo_tool
|
||||
|
||||
|
|
@ -59,7 +59,7 @@ def tearDownClass(cls):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
def test_init(self):
|
||||
# Test normal case.
|
||||
|
|
|
|||
|
|
@ -32,9 +32,9 @@
|
|||
import tuf
|
||||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.sig
|
||||
|
||||
logger = logging.getLogger('tuf.test_sig')
|
||||
|
|
@ -42,7 +42,7 @@
|
|||
# Setup the keys to use in our test cases.
|
||||
KEYS = []
|
||||
for _ in range(3):
|
||||
KEYS.append(tuf.keys.generate_rsa_key(2048))
|
||||
KEYS.append(tuf.ssl_crypto.keys.generate_rsa_key(2048))
|
||||
|
||||
|
||||
|
||||
|
|
@ -52,7 +52,7 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
|
||||
def test_get_signature_status_no_role(self):
|
||||
|
|
@ -76,10 +76,10 @@ def test_get_signature_status_no_role(self):
|
|||
# Should verify we are not adding a duplicate signature
|
||||
# when doing the following action. Here we know 'signable'
|
||||
# has only one signature so it's okay.
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
|
||||
# Improperly formatted role.
|
||||
self.assertRaises(tuf.ssl_commons.exceptions.FormatError, tuf.sig.get_signature_status,
|
||||
|
|
@ -90,17 +90,17 @@ def test_get_signature_status_no_role(self):
|
|||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.sig.verify, *args)
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
|
||||
|
||||
def test_get_signature_status_bad_sig(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
signable['signed'] += 'signature no longer matches signed data'
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
threshold = 1
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid']], threshold)
|
||||
|
|
@ -118,7 +118,7 @@ def test_get_signature_status_bad_sig(self):
|
|||
self.assertFalse(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
# Remove the role.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
||||
|
|
@ -126,11 +126,11 @@ def test_get_signature_status_bad_sig(self):
|
|||
def test_get_signature_status_unknown_method(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
signable['signatures'][0]['method'] = 'fake-sig-method'
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
threshold = 1
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid']], threshold)
|
||||
|
|
@ -149,7 +149,7 @@ def test_get_signature_status_unknown_method(self):
|
|||
self.assertFalse(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
# Remove the role.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
||||
|
|
@ -157,7 +157,7 @@ def test_get_signature_status_unknown_method(self):
|
|||
def test_get_signature_status_single_key(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
threshold = 1
|
||||
|
|
@ -165,7 +165,7 @@ def test_get_signature_status_single_key(self):
|
|||
[KEYS[0]['keyid']], threshold)
|
||||
|
||||
tuf.roledb.add_role('Root', roleinfo)
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
|
||||
sig_status = tuf.sig.get_signature_status(signable, 'Root')
|
||||
|
||||
|
|
@ -189,7 +189,7 @@ def test_get_signature_status_single_key(self):
|
|||
self.assertEqual([], sig_status['unknown_method_sigs'])
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
# Remove the role.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
||||
|
|
@ -197,10 +197,10 @@ def test_get_signature_status_single_key(self):
|
|||
def test_get_signature_status_below_threshold(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
threshold = 2
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid'],
|
||||
|
|
@ -219,7 +219,7 @@ def test_get_signature_status_below_threshold(self):
|
|||
self.assertFalse(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
|
||||
# Remove the role.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
@ -229,13 +229,13 @@ def test_get_signature_status_below_threshold_unrecognized_sigs(self):
|
|||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
# Two keys sign it, but only one of them will be trusted.
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[2], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.keydb.add_key(KEYS[1])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[1])
|
||||
threshold = 2
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid'],
|
||||
|
|
@ -254,8 +254,8 @@ def test_get_signature_status_below_threshold_unrecognized_sigs(self):
|
|||
self.assertFalse(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.keydb.remove_key(KEYS[1]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[1]['keyid'])
|
||||
|
||||
# Remove the role.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
@ -266,13 +266,13 @@ def test_get_signature_status_below_threshold_unauthorized_sigs(self):
|
|||
|
||||
# Two keys sign it, but one of them is only trusted for a different
|
||||
# role.
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[1], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.keydb.add_key(KEYS[1])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[1])
|
||||
threshold = 2
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid'], KEYS[2]['keyid']], threshold)
|
||||
|
|
@ -296,8 +296,8 @@ def test_get_signature_status_below_threshold_unauthorized_sigs(self):
|
|||
tuf.sig.get_signature_status, signable, 'unknown_role')
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.keydb.remove_key(KEYS[1]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[1]['keyid'])
|
||||
|
||||
# Remove the roles.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
@ -308,10 +308,10 @@ def test_get_signature_status_below_threshold_unauthorized_sigs(self):
|
|||
def test_check_signatures_no_role(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
|
||||
# No specific role we're considering. It's invalid to use the
|
||||
# function tuf.sig.verify() without a role specified because
|
||||
|
|
@ -320,16 +320,16 @@ def test_check_signatures_no_role(self):
|
|||
self.assertRaises(tuf.ssl_commons.exceptions.Error, tuf.sig.verify, *args)
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
|
||||
|
||||
|
||||
def test_verify_single_key(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
threshold = 1
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid']], threshold)
|
||||
|
|
@ -340,7 +340,7 @@ def test_verify_single_key(self):
|
|||
self.assertTrue(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
|
||||
# Remove the roles.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
@ -350,13 +350,13 @@ def test_verify_unrecognized_sig(self):
|
|||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
# Two keys sign it, but only one of them will be trusted.
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[2], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[0])
|
||||
tuf.keydb.add_key(KEYS[1])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[0])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[1])
|
||||
threshold = 2
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[0]['keyid'], KEYS[1]['keyid']], threshold)
|
||||
|
|
@ -365,8 +365,8 @@ def test_verify_unrecognized_sig(self):
|
|||
self.assertFalse(tuf.sig.verify(signable, 'Root'))
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.keydb.remove_key(KEYS[1]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[0]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[1]['keyid'])
|
||||
|
||||
# Remove the roles.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
@ -376,7 +376,7 @@ def test_verify_unrecognized_sig(self):
|
|||
def test_generate_rsa_signature(self):
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
self.assertEqual(1, len(signable['signatures']))
|
||||
|
|
@ -386,7 +386,7 @@ def test_generate_rsa_signature(self):
|
|||
returned_signature = tuf.sig.generate_rsa_signature(signable['signed'], KEYS[0])
|
||||
self.assertTrue(tuf.ssl_crypto.formats.SIGNATURE_SCHEMA.matches(returned_signature))
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[1], signable['signed']))
|
||||
|
||||
self.assertEqual(2, len(signable['signatures']))
|
||||
|
|
@ -399,10 +399,10 @@ def test_may_need_new_keys(self):
|
|||
# One untrusted key in 'signable'.
|
||||
signable = {'signed' : 'test', 'signatures' : []}
|
||||
|
||||
signable['signatures'].append(tuf.keys.create_signature(
|
||||
signable['signatures'].append(tuf.ssl_crypto.keys.create_signature(
|
||||
KEYS[0], signable['signed']))
|
||||
|
||||
tuf.keydb.add_key(KEYS[1])
|
||||
tuf.ssl_crypto.keydb.add_key(KEYS[1])
|
||||
threshold = 1
|
||||
roleinfo = tuf.tufformats.make_role_metadata(
|
||||
[KEYS[1]['keyid']], threshold)
|
||||
|
|
@ -414,7 +414,7 @@ def test_may_need_new_keys(self):
|
|||
|
||||
|
||||
# Done. Let's remove the added key(s) from the key database.
|
||||
tuf.keydb.remove_key(KEYS[1]['keyid'])
|
||||
tuf.ssl_crypto.keydb.remove_key(KEYS[1]['keyid'])
|
||||
|
||||
# Remove the roles.
|
||||
tuf.roledb.remove_role('Root')
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@
|
|||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
import tuf.repository_tool as repo_tool
|
||||
import tuf.roledb
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
from simple_settings import settings
|
||||
import six
|
||||
|
||||
|
|
@ -219,7 +219,7 @@ def tearDown(self):
|
|||
# directories that may have been created during each test case.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
def test_with_tuf_mode_1(self):
|
||||
|
|
|
|||
|
|
@ -69,7 +69,7 @@
|
|||
from simple_settings import settings
|
||||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.repository_tool as repo_tool
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
|
|
@ -194,7 +194,7 @@ def tearDown(self):
|
|||
# We are inheriting from custom class.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
@ -332,8 +332,8 @@ def test_1__rebuild_key_and_role_db(self):
|
|||
# keys multiplied by the number of keyid hash algorithms), to include the
|
||||
# delegated targets key. The delegated roles of 'targets.json' are also
|
||||
# loaded when the repository object is instantiated.
|
||||
print('\ndifference: ' + repr(list(set(tuf.keydb._keydb_dict[self.repository_name].keys()) - set(root_metadata['keys'].keys()))))
|
||||
self.assertEqual(number_of_root_keys * 2 + 1, len(tuf.keydb._keydb_dict[self.repository_name]))
|
||||
print('\ndifference: ' + repr(list(set(tuf.ssl_crypto.keydb._keydb_dict[self.repository_name].keys()) - set(root_metadata['keys'].keys()))))
|
||||
self.assertEqual(number_of_root_keys * 2 + 1, len(tuf.ssl_crypto.keydb._keydb_dict[self.repository_name]))
|
||||
|
||||
# Test: normal case.
|
||||
self.repository_updater._rebuild_key_and_role_db()
|
||||
|
|
@ -343,7 +343,7 @@ def test_1__rebuild_key_and_role_db(self):
|
|||
# _rebuild_key_and_role_db() will only rebuild the keys and roles specified
|
||||
# in the 'root.json' file, unlike __init__(). Instantiating an updater
|
||||
# object calls both _rebuild_key_and_role_db() and _import_delegations().
|
||||
self.assertEqual(number_of_root_keys * 2, len(tuf.keydb._keydb_dict[self.repository_name]))
|
||||
self.assertEqual(number_of_root_keys * 2, len(tuf.ssl_crypto.keydb._keydb_dict[self.repository_name]))
|
||||
|
||||
# Test: properly updated roledb and keydb dicts if the Root role changes.
|
||||
root_metadata = self.repository_updater.metadata['current']['root']
|
||||
|
|
@ -354,7 +354,7 @@ def test_1__rebuild_key_and_role_db(self):
|
|||
|
||||
root_roleinfo = tuf.roledb.get_roleinfo('root', self.repository_name)
|
||||
self.assertEqual(root_roleinfo['threshold'], 8)
|
||||
self.assertEqual(number_of_root_keys * 2 - 2, len(tuf.keydb._keydb_dict[self.repository_name]))
|
||||
self.assertEqual(number_of_root_keys * 2 - 2, len(tuf.ssl_crypto.keydb._keydb_dict[self.repository_name]))
|
||||
"""
|
||||
|
||||
|
||||
|
|
@ -459,20 +459,20 @@ def test_2__import_delegations(self):
|
|||
# there without using '_load_metadata_from_file()' since it calls
|
||||
# '_import_delegations()'.
|
||||
repository_name = self.repository_updater.updater_name
|
||||
tuf.keydb.clear_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(repository_name)
|
||||
tuf.roledb.clear_roledb(repository_name)
|
||||
|
||||
self.assertEqual(len(tuf.roledb._roledb_dict[repository_name]), 0)
|
||||
self.assertEqual(len(tuf.keydb._keydb_dict[repository_name]), 0)
|
||||
self.assertEqual(len(tuf.ssl_crypto.keydb._keydb_dict[repository_name]), 0)
|
||||
|
||||
self.repository_updater._rebuild_key_and_role_db()
|
||||
|
||||
self.assertEqual(len(tuf.roledb._roledb_dict[repository_name]), 4)
|
||||
# Take into account the number of keyids algorithms supported by default,
|
||||
# which this test condition expects to be two (sha256 and sha512).
|
||||
print('\nkeydb_dict len: ' + repr(len(tuf.keydb._keydb_dict[repository_name].keys())))
|
||||
print('\nkeydb_dict: ' + repr(tuf.keydb._keydb_dict[repository_name].keys()))
|
||||
self.assertEqual(4 * 2, len(tuf.keydb._keydb_dict[repository_name]))
|
||||
print('\nkeydb_dict len: ' + repr(len(tuf.ssl_crypto.keydb._keydb_dict[repository_name].keys())))
|
||||
print('\nkeydb_dict: ' + repr(tuf.ssl_crypto.keydb._keydb_dict[repository_name].keys()))
|
||||
self.assertEqual(4 * 2, len(tuf.ssl_crypto.keydb._keydb_dict[repository_name]))
|
||||
|
||||
# Test: pass a role without delegations.
|
||||
self.repository_updater._import_delegations('root')
|
||||
|
|
@ -482,7 +482,7 @@ def test_2__import_delegations(self):
|
|||
self.assertEqual(len(tuf.roledb._roledb_dict[repository_name]), 4)
|
||||
# Take into account the number of keyid hash algorithms, which this
|
||||
# test condition expects to be two (for sha256 and sha512).
|
||||
self.assertEqual(len(tuf.keydb._keydb_dict[repository_name]), 4 * 2)
|
||||
self.assertEqual(len(tuf.ssl_crypto.keydb._keydb_dict[repository_name]), 4 * 2)
|
||||
|
||||
# Test: normal case, first level delegation.
|
||||
self.repository_updater._import_delegations('targets')
|
||||
|
|
@ -490,7 +490,7 @@ def test_2__import_delegations(self):
|
|||
self.assertEqual(len(tuf.roledb._roledb_dict[repository_name]), 5)
|
||||
# The number of root keys (times the number of key hash algorithms) +
|
||||
# delegation's key.
|
||||
self.assertEqual(len(tuf.keydb._keydb_dict[repository_name]), 4 * 2 + 1)
|
||||
self.assertEqual(len(tuf.ssl_crypto.keydb._keydb_dict[repository_name]), 4 * 2 + 1)
|
||||
|
||||
# Verify that roledb dictionary was added.
|
||||
self.assertTrue('role1' in tuf.roledb._roledb_dict[repository_name])
|
||||
|
|
@ -504,7 +504,7 @@ def test_2__import_delegations(self):
|
|||
keyids.append(signature['keyid'])
|
||||
|
||||
for keyid in keyids:
|
||||
self.assertTrue(keyid in tuf.keydb._keydb_dict[repository_name])
|
||||
self.assertTrue(keyid in tuf.ssl_crypto.keydb._keydb_dict[repository_name])
|
||||
|
||||
# Verify that _import_delegations() ignores invalid keytypes in the 'keys'
|
||||
# field of parent role's 'delegations'.
|
||||
|
|
@ -520,7 +520,7 @@ def test_2__import_delegations(self):
|
|||
|
||||
# Verify that _import_delegations() raises an exception if any key in
|
||||
# 'delegations' is improperly formatted (i.e., bad keyid).
|
||||
tuf.keydb.clear_keydb(repository_name)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(repository_name)
|
||||
|
||||
self.repository_updater.metadata['current']['targets']['delegations']\
|
||||
['keys'].update({'123': self.repository_updater.metadata['current']\
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@
|
|||
from simple_settings import settings
|
||||
import tuf.log
|
||||
import tuf.tufformats
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.repository_tool as repo_tool
|
||||
import tuf.unittest_toolbox as unittest_toolbox
|
||||
|
|
@ -185,7 +185,7 @@ def tearDown(self):
|
|||
# We are inheriting from custom class.
|
||||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
tuf.roledb.clear_roledb(clear_all=True)
|
||||
tuf.keydb.clear_keydb(clear_all=True)
|
||||
tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -120,8 +120,8 @@
|
|||
import tuf.download
|
||||
import tuf.tufformats
|
||||
import tuf.ssl_crypto.hash
|
||||
import tuf.keys
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.log
|
||||
import tuf.mirrors
|
||||
import tuf.roledb
|
||||
|
|
@ -490,7 +490,7 @@ def _rebuild_key_and_role_db(self):
|
|||
# The metadata files for delegated roles are also not loaded when the
|
||||
# repository is first instantiated. Due to this setup, reloading delegated
|
||||
# roles is not required here.
|
||||
tuf.keydb.create_keydb_from_root_metadata(self.metadata['current']['root'],
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(self.metadata['current']['root'],
|
||||
self.updater_name)
|
||||
tuf.roledb.create_roledb_from_root_metadata(self.metadata['current']['root'],
|
||||
self.updater_name)
|
||||
|
|
@ -538,15 +538,15 @@ def _import_delegations(self, parent_role):
|
|||
# Iterate the keys of the delegated roles of 'parent_role' and load them.
|
||||
for keyid, keyinfo in six.iteritems(keys_info):
|
||||
if keyinfo['keytype'] in ['rsa', 'ed25519']:
|
||||
key, keyids = tuf.keys.format_metadata_to_key(keyinfo)
|
||||
key, keyids = tuf.ssl_crypto.keys.format_metadata_to_key(keyinfo)
|
||||
|
||||
# We specify the keyid to ensure that it's the correct keyid
|
||||
# for the key.
|
||||
try:
|
||||
tuf.keydb.add_key(key, keyid, self.updater_name)
|
||||
tuf.ssl_crypto.keydb.add_key(key, keyid, self.updater_name)
|
||||
for keyid in keyids:
|
||||
key['keyid'] = keyid
|
||||
tuf.keydb.add_key(key, keyid=None, repository_name=self.updater_name)
|
||||
tuf.ssl_crypto.keydb.add_key(key, keyid=None, repository_name=self.updater_name)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -39,9 +39,9 @@
|
|||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.ssl_crypto.util
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.sig
|
||||
import tuf.log
|
||||
from simple_settings import settings
|
||||
|
|
@ -50,8 +50,8 @@
|
|||
|
||||
# These imports provide the interface for 'developer_tool.py', since the imports
|
||||
# are made there.
|
||||
from tuf.keys import format_keyval_to_metadata
|
||||
from tuf.keys import format_metadata_to_key
|
||||
from tuf.ssl_crypto.keys import format_keyval_to_metadata
|
||||
from tuf.ssl_crypto.keys import format_metadata_to_key
|
||||
|
||||
from tuf.repository_tool import Targets
|
||||
from tuf.repository_lib import get_metadata_fileinfo
|
||||
|
|
@ -249,7 +249,7 @@ def write(self, write_partial=False):
|
|||
# Raise 'tuf.ssl_commons.exceptions.FormatError' if any are improperly formatted.
|
||||
tuf.ssl_crypto.formats.BOOLEAN_SCHEMA.check_match(write_partial)
|
||||
|
||||
# At this point the tuf.keydb and tuf.roledb stores must be fully
|
||||
# At this point the tuf.ssl_crypto.keydb and tuf.roledb stores must be fully
|
||||
# populated, otherwise write() throwns a 'tuf.Repository' exception if
|
||||
# any of the project roles are missing signatures, keys, etc.
|
||||
|
||||
|
|
@ -310,7 +310,7 @@ def add_verification_key(self, key):
|
|||
tuf.ssl_commons.exceptions.Error, if the project already contains a key.
|
||||
|
||||
<Side Effects>
|
||||
The role's entries in 'tuf.keydb.py' and 'tuf.roledb.py' are updated.
|
||||
The role's entries in 'tuf.ssl_crypto.keydb.py' and 'tuf.roledb.py' are updated.
|
||||
|
||||
<Returns>
|
||||
None
|
||||
|
|
@ -754,7 +754,7 @@ def _save_project_configuration(metadata_directory, targets_directory,
|
|||
|
||||
# Build a dictionary containing the actual keys.
|
||||
for key in public_keys:
|
||||
key_info = tuf.keydb.get_key(key)
|
||||
key_info = tuf.ssl_crypto.keydb.get_key(key)
|
||||
key_metadata = format_keyval_to_metadata(key_info['keytype'],
|
||||
key_info['keyval'])
|
||||
project_config['public_keys'][key] = key_metadata
|
||||
|
|
@ -807,7 +807,7 @@ def load_project(project_directory, prefix='', new_targets_location=None):
|
|||
|
||||
# Clear the role and key databases since we are loading in a new project.
|
||||
tuf.roledb.clear_roledb()
|
||||
tuf.keydb.clear_keydb()
|
||||
tuf.ssl_crypto.keydb.clear_keydb()
|
||||
|
||||
# Locate metadata filepaths and targets filepath.
|
||||
project_directory = os.path.abspath(project_directory)
|
||||
|
|
@ -857,7 +857,7 @@ def load_project(project_directory, prefix='', new_targets_location=None):
|
|||
keydict = project_configuration['public_keys']
|
||||
|
||||
for keyid in keydict:
|
||||
key, junk = tuf.keys.format_metadata_to_key(keydict[keyid])
|
||||
key, junk = tuf.ssl_crypto.keys.format_metadata_to_key(keydict[keyid])
|
||||
project.add_verification_key(key)
|
||||
|
||||
# Load the project's metadata.
|
||||
|
|
@ -890,8 +890,8 @@ def load_project(project_directory, prefix='', new_targets_location=None):
|
|||
|
||||
|
||||
for key_metadata in targets_metadata['delegations']['keys'].values():
|
||||
key_object, junk = tuf.keys.format_metadata_to_key(key_metadata)
|
||||
tuf.keydb.add_key(key_object)
|
||||
key_object, junk = tuf.ssl_crypto.keys.format_metadata_to_key(key_metadata)
|
||||
tuf.ssl_crypto.keydb.add_key(key_object)
|
||||
|
||||
for role in targets_metadata['delegations']['roles']:
|
||||
rolename = role['name']
|
||||
|
|
@ -972,10 +972,10 @@ def load_project(project_directory, prefix='', new_targets_location=None):
|
|||
|
||||
# Add the keys specified in the delegations field of the Targets role.
|
||||
for key_metadata in metadata_object['delegations']['keys'].values():
|
||||
key_object, junk = tuf.keys.format_metadata_to_key(key_metadata)
|
||||
key_object, junk = tuf.ssl_crypto.keys.format_metadata_to_key(key_metadata)
|
||||
|
||||
try:
|
||||
tuf.keydb.add_key(key_object)
|
||||
tuf.ssl_crypto.keydb.add_key(key_object)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
pass
|
||||
|
|
|
|||
443
tuf/keydb.py
443
tuf/keydb.py
|
|
@ -1,443 +0,0 @@
|
|||
"""
|
||||
<Program Name>
|
||||
keydb.py
|
||||
|
||||
<Author>
|
||||
Vladimir Diaz <vladimir.v.diaz@gmail.com>
|
||||
|
||||
<Started>
|
||||
March 21, 2012. Based on a previous version of this module by Geremy Condra.
|
||||
|
||||
<Copyright>
|
||||
See LICENSE for licensing information.
|
||||
|
||||
<Purpose>
|
||||
Represent a collection of keys and their organization. This module ensures
|
||||
the layout of the collection remain consistent and easily verifiable.
|
||||
Provided are functions to add and delete keys from the database, retrieve a
|
||||
single key, and assemble a collection from keys stored in TUF 'Root' Metadata.
|
||||
The Update Framework process maintains a set of role info for multiple
|
||||
repositories.
|
||||
|
||||
RSA keys are currently supported and a collection of keys is organized as a
|
||||
dictionary indexed by key ID. Key IDs are used as identifiers for keys
|
||||
(e.g., RSA key). They are the hexadecimal representations of the hash of key
|
||||
objects (specifically, the key object containing only the public key). See
|
||||
'rsa_key.py' and the '_get_keyid()' function to learn precisely how keyids
|
||||
are generated. One may get the keyid of a key object by simply accessing the
|
||||
dictionary's 'keyid' key (i.e., rsakey['keyid']).
|
||||
"""
|
||||
|
||||
# Help with Python 3 compatibility, where the print statement is a function, an
|
||||
# implicit relative import is invalid, and the '/' operator performs true
|
||||
# division. Example: print 'hello world' raises a 'SyntaxError' exception.
|
||||
from __future__ import print_function
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import logging
|
||||
import copy
|
||||
|
||||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.keys
|
||||
import six
|
||||
|
||||
# List of strings representing the key types supported by TUF.
|
||||
_SUPPORTED_KEY_TYPES = ['rsa', 'ed25519']
|
||||
|
||||
# See 'log.py' to learn how logging is handled in TUF.
|
||||
logger = logging.getLogger('tuf.keydb')
|
||||
|
||||
# The key database.
|
||||
_keydb_dict = {}
|
||||
_keydb_dict['default'] = {}
|
||||
|
||||
|
||||
def create_keydb_from_root_metadata(root_metadata, repository_name='default'):
|
||||
"""
|
||||
<Purpose>
|
||||
Populate the key database with the unique keys found in 'root_metadata'.
|
||||
The database dictionary will conform to 'tuf.ssl_crypto.formats.KEYDB_SCHEMA' and
|
||||
have the form: {keyid: key, ...}.
|
||||
The 'keyid' conforms to 'tuf.ssl_crypto.formats.KEYID_SCHEMA' and 'key' to its
|
||||
respective type. In the case of RSA keys, this object would match
|
||||
'RSAKEY_SCHEMA'.
|
||||
|
||||
<Arguments>
|
||||
root_metadata:
|
||||
A dictionary conformant to 'tuf.ssl_crypto.formats.ROOT_SCHEMA'. The keys found
|
||||
in the 'keys' field of 'root_metadata' are needed by this function.
|
||||
|
||||
repository_name:
|
||||
The name of the repository to store the key information. If not supplied,
|
||||
the key database is populated for the 'default' repository.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if 'root_metadata' does not have the correct format.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' does not exist in the key
|
||||
database.
|
||||
|
||||
<Side Effects>
|
||||
A function to add the key to the database is called. In the case of RSA
|
||||
keys, this function is add_key().
|
||||
|
||||
The old keydb key database is replaced.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Does 'root_metadata' have the correct format?
|
||||
# This check will ensure 'root_metadata' has the appropriate number of objects
|
||||
# and object types, and that all dict keys are properly named.
|
||||
# Raise 'tuf.ssl_commons.exceptions.FormatError' if the check fails.
|
||||
tuf.ssl_crypto.formats.ROOT_SCHEMA.check_match(root_metadata)
|
||||
|
||||
# Does 'repository_name' have the correct format?
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
# Clear the key database for 'repository_name', or create it if non-existent.
|
||||
if repository_name in _keydb_dict:
|
||||
_keydb_dict[repository_name].clear()
|
||||
|
||||
else:
|
||||
create_keydb(repository_name)
|
||||
|
||||
# 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.
|
||||
for keyid_in_root_metadata, key_metadata in six.iteritems(root_metadata['keys']):
|
||||
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, which is the format expected by 'add_key()'. Note:
|
||||
# The 'keyids' returned by format_metadata_to_key() include keyids in
|
||||
# addition to the default keyid listed in 'key_dict'. The additional
|
||||
# keyids are generated according to settings.REPOSITORY_HASH_ALGORITHMS.
|
||||
key_dict, keyids = tuf.keys.format_metadata_to_key(key_metadata)
|
||||
|
||||
try:
|
||||
for keyid in keyids:
|
||||
# Make sure to update key_dict['keyid'] to use one of the other valid
|
||||
# keyids, otherwise add_key() will have no reference to it.
|
||||
key_dict['keyid'] = keyid
|
||||
add_key(key_dict, keyid=None, repository_name=repository_name)
|
||||
|
||||
# Although keyid duplicates should *not* occur (unique dict keys), log a
|
||||
# warning and continue. Howerver, 'key_dict' may have already been
|
||||
# adding to the keydb elsewhere.
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError as e: # pragma: no cover
|
||||
logger.warning(e)
|
||||
continue
|
||||
|
||||
else:
|
||||
logger.warning('Root Metadata file contains a key with an invalid keytype.')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def create_keydb(repository_name):
|
||||
"""
|
||||
<Purpose>
|
||||
Create a key database for a non-default repository named 'repository_name'.
|
||||
|
||||
<Arguments>
|
||||
repository_name:
|
||||
The name of the repository. An empty key database is created, and keys
|
||||
may be added to via add_key(keyid, repository_name).
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if 'repository_name' is improperly formatted.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' already exists.
|
||||
|
||||
<Side Effects>
|
||||
None.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Is 'repository_name' properly formatted? Raise 'tuf.ssl_commons.exceptions.FormatError' if not.
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
if repository_name in _keydb_dict:
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Repository name already exists:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
_keydb_dict[repository_name] = {}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def remove_keydb(repository_name):
|
||||
"""
|
||||
<Purpose>
|
||||
Remove a key database for a non-default repository named 'repository_name'.
|
||||
The 'default' repository cannot be removed.
|
||||
|
||||
<Arguments>
|
||||
repository_name:
|
||||
The name of the repository to remove. The 'default' repository should
|
||||
not be removed, so 'repository_name' cannot be 'default'.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if 'repository_name' is improperly formatted.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' is 'default'.
|
||||
|
||||
<Side Effects>
|
||||
None.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Is 'repository_name' properly formatted? Raise 'tuf.ssl_commons.exceptions.FormatError' if not.
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
if repository_name not in _keydb_dict:
|
||||
logger.warn('Repository name does not exist: ' + repr(repository_name))
|
||||
return
|
||||
|
||||
if repository_name == 'default':
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Cannot remove the default repository:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
del _keydb_dict[repository_name]
|
||||
|
||||
|
||||
|
||||
|
||||
def add_key(key_dict, keyid=None, repository_name='default'):
|
||||
"""
|
||||
<Purpose>
|
||||
Add 'rsakey_dict' to the key database while avoiding duplicates.
|
||||
If keyid is provided, verify it is the correct keyid for 'rsakey_dict'
|
||||
and raise an exception if it is not.
|
||||
|
||||
<Arguments>
|
||||
key_dict:
|
||||
A dictionary conformant to 'tuf.ssl_crypto.formats.ANYKEY_SCHEMA'.
|
||||
It has the form:
|
||||
|
||||
{'keytype': 'rsa',
|
||||
'keyid': keyid,
|
||||
'keyval': {'public': '-----BEGIN RSA PUBLIC KEY----- ...',
|
||||
'private': '-----BEGIN RSA PRIVATE KEY----- ...'}}
|
||||
|
||||
keyid:
|
||||
An object conformant to 'KEYID_SCHEMA'. It is used as an identifier
|
||||
for RSA keys.
|
||||
|
||||
repository_name:
|
||||
The name of the repository to add the key. If not supplied, the key is
|
||||
added to the 'default' repository.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if the arguments do not have the correct format.
|
||||
|
||||
tuf.ssl_commons.exceptions.Error, if 'keyid' does not match the keyid for 'rsakey_dict'.
|
||||
|
||||
tuf.ssl_commons.exceptions.KeyAlreadyExistsError, if 'rsakey_dict' is found in the key database.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' does not exist in the key
|
||||
database.
|
||||
|
||||
<Side Effects>
|
||||
The keydb key database is modified.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Does 'key_dict' have the correct format?
|
||||
# This check will ensure 'key_dict' has the appropriate number of objects
|
||||
# and object types, and that all dict keys are properly named.
|
||||
# Raise 'tuf.ssl_commons.exceptions.FormatError if the check fails.
|
||||
tuf.ssl_crypto.formats.ANYKEY_SCHEMA.check_match(key_dict)
|
||||
|
||||
# Does 'repository_name' have the correct format?
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
# Does 'keyid' have the correct format?
|
||||
if keyid is not None:
|
||||
# Raise 'tuf.ssl_commons.exceptions.FormatError' if the check fails.
|
||||
tuf.ssl_crypto.formats.KEYID_SCHEMA.check_match(keyid)
|
||||
|
||||
# Check if each keyid found in 'key_dict' matches 'keyid'.
|
||||
if keyid != key_dict['keyid']:
|
||||
raise tuf.ssl_commons.exceptions.Error('Incorrect keyid. Got ' + key_dict['keyid'] + ' but expected ' + keyid)
|
||||
|
||||
# Ensure 'repository_name' is actually set in the key database.
|
||||
if repository_name not in _keydb_dict:
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Repository name does not exist:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
# Check if the keyid belonging to 'key_dict' is not already
|
||||
# available in the key database before returning.
|
||||
keyid = key_dict['keyid']
|
||||
if keyid in _keydb_dict[repository_name]:
|
||||
raise tuf.ssl_commons.exceptions.KeyAlreadyExistsError('Key: ' + keyid)
|
||||
|
||||
_keydb_dict[repository_name][keyid] = copy.deepcopy(key_dict)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def get_key(keyid, repository_name='default'):
|
||||
"""
|
||||
<Purpose>
|
||||
Return the key belonging to 'keyid'.
|
||||
|
||||
<Arguments>
|
||||
keyid:
|
||||
An object conformant to 'tuf.ssl_crypto.formats.KEYID_SCHEMA'. It is used as an
|
||||
identifier for keys.
|
||||
|
||||
repository_name:
|
||||
The name of the repository to get the key. If not supplied, the key is
|
||||
retrieved from the 'default' repository.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if the arguments do not have the correct format.
|
||||
|
||||
tuf.ssl_commons.exceptions.UnknownKeyError, if 'keyid' is not found in the keydb database.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' does not exist in the key
|
||||
database.
|
||||
|
||||
<Side Effects>
|
||||
None.
|
||||
|
||||
<Returns>
|
||||
The key matching 'keyid'. In the case of RSA keys, a dictionary conformant
|
||||
to 'tuf.ssl_crypto.formats.RSAKEY_SCHEMA' is returned.
|
||||
"""
|
||||
|
||||
# Does 'keyid' have the correct format?
|
||||
# This check will ensure 'keyid' has the appropriate number of objects
|
||||
# and object types, and that all dict keys are properly named.
|
||||
# Raise 'tuf.ssl_commons.exceptions.FormatError' is the match fails.
|
||||
tuf.ssl_crypto.formats.KEYID_SCHEMA.check_match(keyid)
|
||||
|
||||
# Does 'repository_name' have the correct format?
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
if repository_name not in _keydb_dict:
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Repository name does not exist:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
# Return the key belonging to 'keyid', if found in the key database.
|
||||
try:
|
||||
return copy.deepcopy(_keydb_dict[repository_name][keyid])
|
||||
|
||||
except KeyError:
|
||||
raise tuf.ssl_commons.exceptions.UnknownKeyError('Key: ' + keyid)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def remove_key(keyid, repository_name='default'):
|
||||
"""
|
||||
<Purpose>
|
||||
Remove the key belonging to 'keyid'.
|
||||
|
||||
<Arguments>
|
||||
keyid:
|
||||
An object conformant to 'tuf.ssl_crypto.formats.KEYID_SCHEMA'. It is used as an
|
||||
identifier for keys.
|
||||
|
||||
repository_name:
|
||||
The name of the repository to remove the key. If not supplied, the key
|
||||
is removed from the 'default' repository.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if the arguments do not have the correct format.
|
||||
|
||||
tuf.ssl_commons.exceptions.UnknownKeyError, if 'keyid' is not found in key database.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' does not exist in the key
|
||||
database.
|
||||
|
||||
<Side Effects>
|
||||
The key, identified by 'keyid', is deleted from the key database.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Does 'keyid' have the correct format?
|
||||
# This check will ensure 'keyid' has the appropriate number of objects
|
||||
# and object types, and that all dict keys are properly named.
|
||||
# Raise 'tuf.ssl_commons.exceptions.FormatError' is the match fails.
|
||||
tuf.ssl_crypto.formats.KEYID_SCHEMA.check_match(keyid)
|
||||
|
||||
# Does 'repository_name' have the correct format?
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
|
||||
if repository_name not in _keydb_dict:
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Repository name does not exist:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
# Remove the key belonging to 'keyid' if found in the key database.
|
||||
if keyid in _keydb_dict[repository_name]:
|
||||
del _keydb_dict[repository_name][keyid]
|
||||
|
||||
else:
|
||||
raise tuf.ssl_commons.exceptions.UnknownKeyError('Key: ' + keyid)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def clear_keydb(repository_name='default', clear_all=False):
|
||||
|
||||
"""
|
||||
<Purpose>
|
||||
Clear the keydb key database.
|
||||
|
||||
<Arguments>
|
||||
repository_name:
|
||||
The name of the repository to clear the key database. If not supplied,
|
||||
the key database is cleared for the 'default' repository.
|
||||
|
||||
clear_all:
|
||||
Boolean indicating whether to clear the entire keydb.
|
||||
|
||||
<Exceptions>
|
||||
tuf.ssl_commons.exceptions.FormatError, if 'repository_name' is improperly formatted.
|
||||
|
||||
tuf.ssl_commons.exceptions.InvalidNameError, if 'repository_name' does not exist in the key
|
||||
database.
|
||||
|
||||
<Side Effects>
|
||||
The keydb key database is reset.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
"""
|
||||
|
||||
# Do the arguments have the correct format? Raise 'tuf.ssl_commons.exceptions.FormatError' if
|
||||
# 'repository_name' is improperly formatted.
|
||||
tuf.ssl_crypto.formats.NAME_SCHEMA.check_match(repository_name)
|
||||
tuf.ssl_crypto.formats.BOOLEAN_SCHEMA.check_match(clear_all)
|
||||
|
||||
global _keydb_dict
|
||||
|
||||
if clear_all:
|
||||
_keydb_dict = {}
|
||||
_keydb_dict['default'] = {}
|
||||
|
||||
if repository_name not in _keydb_dict:
|
||||
raise tuf.ssl_commons.exceptions.InvalidNameError('Repository name does not exist:'
|
||||
' ' + repr(repository_name))
|
||||
|
||||
_keydb_dict[repository_name] = {}
|
||||
1486
tuf/keys.py
1486
tuf/keys.py
File diff suppressed because it is too large
Load diff
|
|
@ -45,9 +45,9 @@
|
|||
import tuf.ssl_crypto.formats
|
||||
import tuf.tufformats
|
||||
import tuf.ssl_crypto.util
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.sig
|
||||
import tuf.log
|
||||
from simple_settings import settings
|
||||
|
|
@ -423,16 +423,16 @@ def _remove_invalid_and_duplicate_signatures(signable):
|
|||
key = None
|
||||
|
||||
# Remove 'signature' from 'signable' if the listed keyid does not exist
|
||||
# in 'tuf.keydb'.
|
||||
# in 'tuf.ssl_crypto.keydb'.
|
||||
try:
|
||||
key = tuf.keydb.get_key(keyid)
|
||||
key = tuf.ssl_crypto.keydb.get_key(keyid)
|
||||
|
||||
except tuf.ssl_commons.exceptions.UnknownKeyError:
|
||||
signable['signatures'].remove(signature)
|
||||
continue
|
||||
|
||||
# Remove 'signature' from 'signable' if it is an invalid signature.
|
||||
if not tuf.keys.verify_signature(key, signature, signed):
|
||||
if not tuf.ssl_crypto.keys.verify_signature(key, signature, signed):
|
||||
logger.debug('Removing invalid signature for ' + repr(keyid))
|
||||
signable['signatures'].remove(signature)
|
||||
|
||||
|
|
@ -610,7 +610,7 @@ def _load_top_level_metadata(repository, top_level_filenames):
|
|||
signable = tuf.ssl_crypto.util.load_json_file(root_filename)
|
||||
tuf.tufformats.check_signable_object_format(signable)
|
||||
root_metadata = signable['signed']
|
||||
tuf.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
tuf.ssl_crypto.keydb.create_keydb_from_root_metadata(root_metadata)
|
||||
tuf.roledb.create_roledb_from_root_metadata(root_metadata)
|
||||
|
||||
# Load Root's roleinfo and update 'tuf.roledb'.
|
||||
|
|
@ -768,7 +768,7 @@ def _load_top_level_metadata(repository, top_level_filenames):
|
|||
|
||||
# Add the keys specified in the delegations field of the Targets role.
|
||||
for key_metadata in six.itervalues(targets_metadata['delegations']['keys']):
|
||||
key_object, keyids = tuf.keys.format_metadata_to_key(key_metadata)
|
||||
key_object, keyids = tuf.ssl_crypto.keys.format_metadata_to_key(key_metadata)
|
||||
|
||||
# Add 'key_object' to the list of recognized keys. Keys may be shared,
|
||||
# so do not raise an exception if 'key_object' has already been loaded.
|
||||
|
|
@ -777,10 +777,10 @@ def _load_top_level_metadata(repository, top_level_filenames):
|
|||
# repository maintainer should have also been made aware of the duplicate
|
||||
# key when it was added.
|
||||
try:
|
||||
tuf.keydb.add_key(key_object)
|
||||
tuf.ssl_crypto.keydb.add_key(key_object)
|
||||
for keyid in keyids: #pragma: no branch
|
||||
key_object['keyid'] = keyid
|
||||
tuf.keydb.add_key(key_object, keyid=None)
|
||||
tuf.ssl_crypto.keydb.add_key(key_object, keyid=None)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
pass
|
||||
|
|
@ -873,10 +873,10 @@ def generate_and_write_rsa_keypair(filepath, bits=DEFAULT_RSA_KEY_BITS,
|
|||
|
||||
# Generate public and private RSA keys, encrypted the private portion
|
||||
# and store them in PEM format.
|
||||
rsa_key = tuf.keys.generate_rsa_key(bits)
|
||||
rsa_key = tuf.ssl_crypto.keys.generate_rsa_key(bits)
|
||||
public = rsa_key['keyval']['public']
|
||||
private = rsa_key['keyval']['private']
|
||||
encrypted_pem = tuf.keys.create_rsa_encrypted_pem(private, password)
|
||||
encrypted_pem = tuf.ssl_crypto.keys.create_rsa_encrypted_pem(private, password)
|
||||
|
||||
# Write public key (i.e., 'public', which is in PEM format) to
|
||||
# '<filepath>.pub'. If the parent directory of filepath does not exist,
|
||||
|
|
@ -959,7 +959,7 @@ def import_rsa_privatekey_from_file(filepath, password=None):
|
|||
|
||||
# Convert 'encrypted_pem' to 'tuf.ssl_crypto.formats.RSAKEY_SCHEMA' format. Raise
|
||||
# 'tuf.ssl_commons.exceptions.CryptoError' if 'encrypted_pem' is invalid.
|
||||
rsa_key = tuf.keys.import_rsakey_from_encrypted_pem(encrypted_pem, password)
|
||||
rsa_key = tuf.ssl_crypto.keys.import_rsakey_from_encrypted_pem(encrypted_pem, password)
|
||||
|
||||
return rsa_key
|
||||
|
||||
|
|
@ -1009,7 +1009,7 @@ def import_rsa_publickey_from_file(filepath):
|
|||
|
||||
# Convert 'rsa_pubkey_pem' to 'tuf.ssl_crypto.formats.RSAKEY_SCHEMA' format.
|
||||
try:
|
||||
rsakey_dict = tuf.keys.format_rsakey_from_pem(rsa_pubkey_pem)
|
||||
rsakey_dict = tuf.ssl_crypto.keys.format_rsakey_from_pem(rsa_pubkey_pem)
|
||||
|
||||
except tuf.ssl_commons.exceptions.FormatError as e:
|
||||
raise tuf.ssl_commons.exceptions.Error('Cannot import improperly formatted PEM file.' + repr(str(e)))
|
||||
|
|
@ -1076,15 +1076,15 @@ def generate_and_write_ed25519_keypair(filepath, password=None):
|
|||
# used is determined by the user, or by default (set in
|
||||
# 'settings.ED25519_CRYPTO_LIBRARY'). Raise 'tuf.ssl_commons.exceptions.CryptoError' or
|
||||
# 'tuf.ssl_commons.exceptions.UnsupportedLibraryError', if 'ed25519_key' cannot be encrypted.
|
||||
ed25519_key = tuf.keys.generate_ed25519_key()
|
||||
encrypted_key = tuf.keys.encrypt_key(ed25519_key, password)
|
||||
ed25519_key = tuf.ssl_crypto.keys.generate_ed25519_key()
|
||||
encrypted_key = tuf.ssl_crypto.keys.encrypt_key(ed25519_key, password)
|
||||
|
||||
# ed25519 public key file contents in metadata format (i.e., does not include
|
||||
# the keyid portion).
|
||||
keytype = ed25519_key['keytype']
|
||||
keyval = ed25519_key['keyval']
|
||||
ed25519key_metadata_format = \
|
||||
tuf.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
tuf.ssl_crypto.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
|
||||
# Write the public key, conformant to 'tuf.ssl_crypto.formats.KEY_SCHEMA', to
|
||||
# '<filepath>.pub'.
|
||||
|
|
@ -1142,10 +1142,10 @@ def import_ed25519_publickey_from_file(filepath):
|
|||
# loaded key object in tuf.ssl_crypto.formats.ED25519KEY_SCHEMA' format that also
|
||||
# includes the keyid.
|
||||
ed25519_key_metadata = tuf.ssl_crypto.util.load_json_file(filepath)
|
||||
ed25519_key, junk = tuf.keys.format_metadata_to_key(ed25519_key_metadata)
|
||||
ed25519_key, junk = tuf.ssl_crypto.keys.format_metadata_to_key(ed25519_key_metadata)
|
||||
|
||||
# Raise an exception if an unexpected key type is imported.
|
||||
# Redundant validation of 'keytype'. 'tuf.keys.format_metadata_to_key()'
|
||||
# Redundant validation of 'keytype'. 'tuf.ssl_crypto.keys.format_metadata_to_key()'
|
||||
# should have fully validated 'ed25519_key_metadata'.
|
||||
if ed25519_key['keytype'] != 'ed25519': # pragma: no cover
|
||||
message = 'Invalid key type loaded: ' + repr(ed25519_key['keytype'])
|
||||
|
|
@ -1223,7 +1223,7 @@ def import_ed25519_privatekey_from_file(filepath, password=None):
|
|||
# (i.e., set by the user) and generating the derived encryption key from
|
||||
# 'password'. Raise 'tuf.ssl_commons.exceptions.CryptoError' or 'tuf.ssl_commons.exceptions.UnsupportedLibraryError' if the
|
||||
# decryption fails.
|
||||
key_object = tuf.keys.decrypt_key(encrypted_key, password)
|
||||
key_object = tuf.ssl_crypto.keys.decrypt_key(encrypted_key, password)
|
||||
|
||||
# Raise an exception if an unexpected key type is imported.
|
||||
if key_object['keytype'] != 'ed25519':
|
||||
|
|
@ -1435,7 +1435,7 @@ def generate_root_metadata(version, expiration_date, consistent_snapshot,
|
|||
compression_algorithms=['gz']):
|
||||
"""
|
||||
<Purpose>
|
||||
Create the root metadata. 'tuf.roledb.py' and 'tuf.keydb.py' are read and
|
||||
Create the root metadata. 'tuf.roledb.py' and 'tuf.ssl_crypto.keydb.py' are read and
|
||||
the information returned by these modules is used to generate the root
|
||||
metadata object.
|
||||
|
||||
|
|
@ -1467,7 +1467,7 @@ def generate_root_metadata(version, expiration_date, consistent_snapshot,
|
|||
metadata object (e.g., a required top-level role not found in 'tuf.roledb'.)
|
||||
|
||||
<Side Effects>
|
||||
The contents of 'tuf.keydb.py' and 'tuf.roledb.py' are read.
|
||||
The contents of 'tuf.ssl_crypto.keydb.py' and 'tuf.roledb.py' are read.
|
||||
|
||||
<Returns>
|
||||
A root metadata object, conformant to 'tuf.ssl_crypto.formats.ROOT_SCHEMA'.
|
||||
|
|
@ -1501,7 +1501,7 @@ def generate_root_metadata(version, expiration_date, consistent_snapshot,
|
|||
|
||||
# Generate keys for the keyids listed by the role being processed.
|
||||
for keyid in tuf.roledb.get_role_keyids(rolename):
|
||||
key = tuf.keydb.get_key(keyid)
|
||||
key = tuf.ssl_crypto.keydb.get_key(keyid)
|
||||
|
||||
# If 'key' is an RSA key, it would conform to 'tuf.ssl_crypto.formats.RSAKEY_SCHEMA',
|
||||
# and have the form:
|
||||
|
|
@ -1517,7 +1517,7 @@ def generate_root_metadata(version, expiration_date, consistent_snapshot,
|
|||
keytype = key['keytype']
|
||||
keyval = key['keyval']
|
||||
keydict[keyid] = \
|
||||
tuf.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
tuf.ssl_crypto.keys.format_keyval_to_metadata(keytype, keyval, private=False)
|
||||
|
||||
# This is not a recognized key. Raise an exception.
|
||||
else:
|
||||
|
|
@ -1857,7 +1857,7 @@ def sign_metadata(metadata_object, keyids, filename):
|
|||
<Purpose>
|
||||
Sign a metadata object. If any of the keyids have already signed the file,
|
||||
the old signature is replaced. The keys in 'keyids' must already be
|
||||
loaded in 'tuf.keydb'.
|
||||
loaded in 'tuf.ssl_crypto.keydb'.
|
||||
|
||||
<Arguments>
|
||||
metadata_object:
|
||||
|
|
@ -1904,14 +1904,14 @@ def sign_metadata(metadata_object, keyids, filename):
|
|||
for keyid in keyids:
|
||||
|
||||
# Load the signing key.
|
||||
key = tuf.keydb.get_key(keyid)
|
||||
key = tuf.ssl_crypto.keydb.get_key(keyid)
|
||||
|
||||
# Generate the signature using the appropriate signing method.
|
||||
if key['keytype'] in SUPPORTED_KEY_TYPES:
|
||||
if 'private' in key['keyval']:
|
||||
signed = signable['signed']
|
||||
try:
|
||||
signature = tuf.keys.create_signature(key, signed)
|
||||
signature = tuf.ssl_crypto.keys.create_signature(key, signed)
|
||||
signable['signatures'].append(signature)
|
||||
|
||||
except Exception:
|
||||
|
|
|
|||
|
|
@ -41,9 +41,9 @@
|
|||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.ssl_crypto.util
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
import tuf.keys
|
||||
import tuf.ssl_crypto.keys
|
||||
import tuf.sig
|
||||
import tuf.log
|
||||
import tuf.repository_lib as repo_lib
|
||||
|
|
@ -100,7 +100,7 @@
|
|||
TIMESTAMP_EXPIRATION = 86400
|
||||
|
||||
try:
|
||||
tuf.keys.check_crypto_libraries(['rsa', 'ed25519', 'general'])
|
||||
tuf.ssl_crypto.keys.check_crypto_libraries(['rsa', 'ed25519', 'general'])
|
||||
|
||||
except tuf.ssl_commons.exceptions.UnsupportedLibraryError: #pragma: no cover
|
||||
logger.warn('Warning: The repository and developer tools require'
|
||||
|
|
@ -217,7 +217,7 @@ def writeall(self, consistent_snapshot=False, compression_algorithms=['gz']):
|
|||
tuf.ssl_crypto.formats.BOOLEAN_SCHEMA.check_match(consistent_snapshot)
|
||||
tuf.ssl_crypto.formats.COMPRESSIONS_SCHEMA.check_match(compression_algorithms)
|
||||
|
||||
# At this point, tuf.keydb and tuf.roledb must be fully populated,
|
||||
# At this point, tuf.ssl_crypto.keydb and tuf.roledb must be fully populated,
|
||||
# otherwise writeall() throws a 'tuf.ssl_commons.exceptions.UnsignedMetadataError' for the
|
||||
# top-level roles. exception if any of the top-level roles are missing
|
||||
# signatures, keys, etc.
|
||||
|
|
@ -595,7 +595,7 @@ def add_verification_key(self, key, expires=None):
|
|||
tuf.ssl_commons.exceptions.Error, if the 'expires' datetime has already expired.
|
||||
|
||||
<Side Effects>
|
||||
The role's entries in 'tuf.keydb.py' and 'tuf.roledb.py' are updated.
|
||||
The role's entries in 'tuf.ssl_crypto.keydb.py' and 'tuf.roledb.py' are updated.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
|
|
@ -653,10 +653,10 @@ def add_verification_key(self, key, expires=None):
|
|||
key['expires'] = expires
|
||||
|
||||
# Ensure 'key', which should contain the public portion, is added to
|
||||
# 'tuf.keydb.py'. Add 'key' to the list of recognized keys. Keys may be
|
||||
# 'tuf.ssl_crypto.keydb.py'. Add 'key' to the list of recognized keys. Keys may be
|
||||
# shared, so do not raise an exception if 'key' has already been loaded.
|
||||
try:
|
||||
tuf.keydb.add_key(key)
|
||||
tuf.ssl_crypto.keydb.add_key(key)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
logger.warning('Adding a verification key that has already been used.')
|
||||
|
|
@ -747,7 +747,7 @@ def load_signing_key(self, key):
|
|||
tuf.ssl_commons.exceptions.Error, if the private key is not found in 'key'.
|
||||
|
||||
<Side Effects>
|
||||
Updates the role's 'tuf.keydb.py' and 'tuf.roledb.py' entries.
|
||||
Updates the role's 'tuf.ssl_crypto.keydb.py' and 'tuf.roledb.py' entries.
|
||||
|
||||
<Returns>
|
||||
None.
|
||||
|
|
@ -767,11 +767,11 @@ def load_signing_key(self, key):
|
|||
# Has the key, with the private portion included, been added to the keydb?
|
||||
# The public version of the key may have been previously added.
|
||||
try:
|
||||
tuf.keydb.add_key(key)
|
||||
tuf.ssl_crypto.keydb.add_key(key)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
tuf.keydb.remove_key(key['keyid'])
|
||||
tuf.keydb.add_key(key)
|
||||
tuf.ssl_crypto.keydb.remove_key(key['keyid'])
|
||||
tuf.ssl_crypto.keydb.add_key(key)
|
||||
|
||||
# Update the role's 'signing_keys' field in 'tuf.roledb.py'.
|
||||
roleinfo = tuf.roledb.get_roleinfo(self.rolename)
|
||||
|
|
@ -2180,7 +2180,7 @@ def delegate(self, rolename, public_keys, list_of_targets, threshold=1,
|
|||
|
||||
<Side Effects>
|
||||
A new Target object is created for 'rolename' that is accessible to the
|
||||
caller (i.e., targets.<rolename>). The 'tuf.keydb.py' and
|
||||
caller (i.e., targets.<rolename>). The 'tuf.ssl_crypto.keydb.py' and
|
||||
'tuf.roledb.py' stores are updated with 'public_keys'.
|
||||
|
||||
<Returns>
|
||||
|
|
@ -2212,10 +2212,10 @@ def delegate(self, rolename, public_keys, list_of_targets, threshold=1,
|
|||
keyids = []
|
||||
keydict = {}
|
||||
|
||||
# Add all the keys in 'public_keys' to tuf.keydb.
|
||||
# Add all the keys in 'public_keys' to tuf.ssl_crypto.keydb.
|
||||
for key in public_keys:
|
||||
keyid = key['keyid']
|
||||
key_metadata_format = tuf.keys.format_keyval_to_metadata(key['keytype'],
|
||||
key_metadata_format = tuf.ssl_crypto.keys.format_keyval_to_metadata(key['keytype'],
|
||||
key['keyval'])
|
||||
# Update 'keyids' and 'keydict'.
|
||||
new_keydict = {keyid: key_metadata_format}
|
||||
|
|
@ -2869,7 +2869,7 @@ def load_repository(repository_directory):
|
|||
|
||||
# Load top-level metadata.
|
||||
#tuf.roledb.clear_roledb(clear_all=True)
|
||||
#tuf.keydb.clear_keydb(clear_all=True)
|
||||
#tuf.ssl_crypto.keydb.clear_keydb(clear_all=True)
|
||||
|
||||
repository_directory = os.path.abspath(repository_directory)
|
||||
metadata_directory = os.path.join(repository_directory,
|
||||
|
|
@ -2992,9 +2992,9 @@ def load_repository(repository_directory):
|
|||
# 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']):
|
||||
key_object, junk = tuf.keys.format_metadata_to_key(key_metadata)
|
||||
key_object, junk = tuf.ssl_crypto.keys.format_metadata_to_key(key_metadata)
|
||||
try:
|
||||
tuf.keydb.add_key(key_object)
|
||||
tuf.ssl_crypto.keydb.add_key(key_object)
|
||||
|
||||
except tuf.ssl_commons.exceptions.KeyAlreadyExistsError:
|
||||
pass
|
||||
|
|
|
|||
12
tuf/sig.py
12
tuf/sig.py
|
|
@ -47,7 +47,7 @@
|
|||
|
||||
import tuf
|
||||
import tuf.tufformats
|
||||
import tuf.keydb
|
||||
import tuf.ssl_crypto.keydb
|
||||
import tuf.roledb
|
||||
|
||||
|
||||
|
|
@ -66,7 +66,7 @@ def get_signature_status(signable, role=None, repository_name='default',
|
|||
<Purpose>
|
||||
Return a dictionary representing the status of the signatures listed in
|
||||
'signable'. Given an object conformant to SIGNABLE_SCHEMA, a set of public
|
||||
keys in 'tuf.keydb', a set of roles in 'tuf.roledb', and a role, the status
|
||||
keys in 'tuf.ssl_crypto.keydb', a set of roles in 'tuf.roledb', and a role, the status
|
||||
of these signatures can be determined. This method will iterate the
|
||||
signatures in 'signable' and enumerate all the keys that are valid,
|
||||
invalid, unrecognized, unauthorized, or generated using an unknown method.
|
||||
|
|
@ -158,7 +158,7 @@ def get_signature_status(signable, role=None, repository_name='default',
|
|||
|
||||
# Does the signature use an unrecognized key?
|
||||
try:
|
||||
key = tuf.keydb.get_key(keyid, repository_name)
|
||||
key = tuf.ssl_crypto.keydb.get_key(keyid, repository_name)
|
||||
|
||||
except tuf.ssl_commons.exceptions.UnknownKeyError:
|
||||
unknown_sigs.append(keyid)
|
||||
|
|
@ -166,7 +166,7 @@ def get_signature_status(signable, role=None, repository_name='default',
|
|||
|
||||
# Does the signature use an unknown key signing method?
|
||||
try:
|
||||
valid_sig = tuf.keys.verify_signature(key, signature, signed)
|
||||
valid_sig = tuf.ssl_crypto.keys.verify_signature(key, signature, signed)
|
||||
|
||||
except tuf.ssl_commons.exceptions.UnknownMethodError:
|
||||
unknown_method_sigs.append(keyid)
|
||||
|
|
@ -354,7 +354,7 @@ def generate_rsa_signature(signed, rsakey_dict):
|
|||
|
||||
<Arguments>
|
||||
signed:
|
||||
The data used by 'tuf.keys.create_signature()' to generate signatures.
|
||||
The data used by 'tuf.ssl_crypto.keys.create_signature()' to generate signatures.
|
||||
It is stored in the 'signed' field of 'signable'.
|
||||
|
||||
rsakey_dict:
|
||||
|
|
@ -381,6 +381,6 @@ def generate_rsa_signature(signed, rsakey_dict):
|
|||
|
||||
# Generate the RSA signature.
|
||||
# Raises tuf.ssl_commons.exceptions.FormatError and TypeError.
|
||||
signature = tuf.keys.create_signature(rsakey_dict, signed)
|
||||
signature = tuf.ssl_crypto.keys.create_signature(rsakey_dict, signed)
|
||||
|
||||
return signature
|
||||
|
|
|
|||
Loading…
Reference in a new issue