Port to new securesystemslib w abstract filesystem

Switch to using the new abstract files and directories support in
securesystemslib by taking an object which implements
securesystemslib.storage.StorageBackendInterface in the Repository
constructor, passed in by tuf.repository_tool.create_new_repository() and
tuf.repository_tool.load_repository()

The Updater class in tuf.client.updater does not specify a storage backend
and instead allows the functions in securesystemslib to perform the
default action of instantiating a LocalFilesystemBackend, that is the
updater does not currently support abstract filesystem backends and always
defaults to using local storage.

Finall we drop support for tuf.settings.CONSISTENT_METHOD as it's not as
clear how different copying modes should work when the details of the
underlying storage are abstracted away.

Signed-off-by: Joshua Lock <[email protected]>

Author: joshuagl

tests/test_developer_tool.py

@@ -38,6 +38,7 @@
 import tuf.exceptions
 
 import securesystemslib
+import securesystemslib.exceptions
 
 from tuf.developer_tool import METADATA_DIRECTORY_NAME
 from tuf.developer_tool import TARGETS_DIRECTORY_NAME
@@ -188,7 +189,8 @@ def test_load_project(self):
 
     # Test non-existent project filepath.
     nonexistent_path = os.path.join(local_tmp, 'nonexistent')
-    self.assertRaises(IOError, developer_tool.load_project, nonexistent_path)
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+        developer_tool.load_project, nonexistent_path)
 
     # Copy the pregenerated metadata.
     project_data_filepath = os.path.join('repository_data', 'project')

tests/test_multiple_repositories_integration.py

@@ -164,7 +164,7 @@ def setUp(self):
                                            'targets_path': 'targets',
                                            'confined_target_dirs': ['']}}
 
-    self.repository_mirrors2 = {'mirror1': {'url_prefix': url_prefix2,
+    self.repository_mirrors2 = {'mirror2': {'url_prefix': url_prefix2,
                                            'metadata_path': 'metadata',
                                            'targets_path': 'targets',
                                            'confined_target_dirs': ['']}}

tests/test_repository_lib.py

@@ -52,8 +52,10 @@
 import tuf.repository_tool as repo_tool
 
 import securesystemslib
+import securesystemslib.exceptions
 import securesystemslib.rsa_keys
 import securesystemslib.interface
+import securesystemslib.storage
 import six
 
 logger = logging.getLogger(__name__)
@@ -126,8 +128,9 @@ def test_import_rsa_privatekey_from_file(self):
     # Non-existent key file.
     nonexistent_keypath = os.path.join(temporary_directory,
                                        'nonexistent_keypath')
-    self.assertRaises(IOError, repo_lib.import_rsa_privatekey_from_file,
-                      nonexistent_keypath, 'pw')
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+        repo_lib.import_rsa_privatekey_from_file,
+        nonexistent_keypath, 'pw')
 
     # Invalid key file argument.
     invalid_keyfile = os.path.join(temporary_directory, 'invalid_keyfile')
@@ -160,7 +163,8 @@ def test_import_ed25519_privatekey_from_file(self):
     # Non-existent key file.
     nonexistent_keypath = os.path.join(temporary_directory,
                                        'nonexistent_keypath')
-    self.assertRaises(IOError, repo_lib.import_ed25519_privatekey_from_file,
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+                      repo_lib.import_ed25519_privatekey_from_file,
                       nonexistent_keypath, 'pw')
 
     # Invalid key file argument.
@@ -215,7 +219,7 @@ def test_get_metadata_filenames(self):
                  'targets.json': os.path.join(metadata_directory, 'targets.json'),
                  'snapshot.json': os.path.join(metadata_directory, 'snapshot.json'),
                  'timestamp.json': os.path.join(metadata_directory, 'timestamp.json')}
-    self.assertEqual(filenames, repo_lib.get_metadata_filenames())
+    self.assertEqual(filenames, repo_lib.get_metadata_filenames(metadata_directory))
 
 
     # Test improperly formatted argument.
@@ -440,8 +444,9 @@ def test_generate_snapshot_metadata(self):
 
     # Load a valid repository so that top-level roles exist in roledb and
     # generate_snapshot_metadata() has roles to specify in snapshot metadata.
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     repository = repo_tool.Repository(repository_directory, metadata_directory,
-                                      targets_directory)
+                                      targets_directory, storage_backend)
 
     repository_junk = repo_tool.load_repository(repository_directory)
 
@@ -458,26 +463,27 @@ def test_generate_snapshot_metadata(self):
       repo_lib.generate_snapshot_metadata(metadata_directory, version,
                                           expiration_date,
                                           targets_filename,
+                                          storage_backend,
                                           consistent_snapshot=False)
     self.assertTrue(tuf.formats.SNAPSHOT_SCHEMA.matches(snapshot_metadata))
 
 
     # Test improperly formatted arguments.
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.generate_snapshot_metadata,
                       3, version, expiration_date,
-                      targets_filename, consistent_snapshot=False)
+                      targets_filename, consistent_snapshot=False, storage_backend=storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.generate_snapshot_metadata,
                       metadata_directory, '3', expiration_date,
-                      targets_filename, consistent_snapshot=False)
+                      targets_filename, storage_backend, consistent_snapshot=False)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.generate_snapshot_metadata,
                       metadata_directory, version, '3',
-                      targets_filename, consistent_snapshot=False)
+                      targets_filename, storage_backend, consistent_snapshot=False)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.generate_snapshot_metadata,
                       metadata_directory, version, expiration_date,
-                      3, consistent_snapshot=False)
+                      3, storage_backend, consistent_snapshot=False)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.generate_snapshot_metadata,
                       metadata_directory, version, expiration_date,
-                      targets_filename, 3)
+                      targets_filename, 3, storage_backend)
 
 
 
@@ -599,85 +605,25 @@ def test_write_metadata_file(self):
     version_number = root_signable['signed']['version'] + 1
 
     self.assertFalse(os.path.exists(output_filename))
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     repo_lib.write_metadata_file(root_signable, output_filename, version_number,
-        consistent_snapshot=False)
+        consistent_snapshot=False, storage_backend=storage_backend)
     self.assertTrue(os.path.exists(output_filename))
 
     # Attempt to over-write the previously written metadata file.  An exception
     # is not raised in this case, only a debug message is logged.
     repo_lib.write_metadata_file(root_signable, output_filename, version_number,
-        consistent_snapshot=False)
-
-    # Try to write a consistent metadate file. An exception is not raised in
-    # this case.  For testing purposes, root.json should be a hard link to the
-    # consistent metadata file.  We should verify that root.json points to
-    # the latest consistent files.
-    tuf.settings.CONSISTENT_METHOD = 'hard_link'
-    repo_lib.write_metadata_file(root_signable, output_filename, version_number,
-        consistent_snapshot=True)
-
-    # Test if the consistent files are properly named
-    # Filename format of a consistent file: <version number>.rolename.json
-    version_and_filename = str(version_number) + '.' + 'root.json'
-    first_version_output_file = os.path.join(temporary_directory, version_and_filename)
-    self.assertTrue(os.path.exists(first_version_output_file))
-
-    # Verify that the consistent file content is equal to 'output_filename'.
-    self.assertEqual(
-        securesystemslib.util.get_file_details(output_filename),
-        securesystemslib.util.get_file_details(first_version_output_file))
-
-    # Try to add more consistent metadata files.
-    version_number += 1
-    root_signable['signed']['version'] = version_number
-    repo_lib.write_metadata_file(root_signable, output_filename,
-        version_number, consistent_snapshot=True)
-
-    # Test if the latest root.json points to the expected consistent file
-    # and consistent metadata do not all point to the same root.json
-    version_and_filename = str(version_number) + '.' + 'root.json'
-    second_version_output_file = os.path.join(temporary_directory, version_and_filename)
-    self.assertTrue(os.path.exists(second_version_output_file))
-
-    # Verify that the second version is equal to the second output file, and
-    # that the second output filename differs from the first.
-    self.assertEqual(securesystemslib.util.get_file_details(output_filename),
-        securesystemslib.util.get_file_details(second_version_output_file))
-    self.assertNotEqual(securesystemslib.util.get_file_details(output_filename),
-        securesystemslib.util.get_file_details(first_version_output_file))
-
-    # Test for an improper settings.CONSISTENT_METHOD string value.
-    tuf.settings.CONSISTENT_METHOD = 'somebadidea'
-
-    # Test for invalid consistent methods on systems other than Windows,
-    # which always uses the copy method.
-    if platform.system() == 'Windows':
-      pass
-
-    else:
-      self.assertRaises(securesystemslib.exceptions.InvalidConfigurationError,
-          repo_lib.write_metadata_file, root_signable, output_filename,
-          version_number, consistent_snapshot=True)
-
-    # Try to create a link to root.json when root.json doesn't exist locally.
-    # repository_lib should log a message if this is the case.
-    tuf.settings.CONSISTENT_METHOD = 'hard_link'
-    os.remove(output_filename)
-    repo_lib.write_metadata_file(root_signable, output_filename, version_number,
-        consistent_snapshot=True)
-
-    # Reset CONSISTENT_METHOD so that subsequent tests work as expected.
-    tuf.settings.CONSISTENT_METHOD = 'copy'
+        consistent_snapshot=False, storage_backend=storage_backend)
 
     # Test improperly formatted arguments.
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.write_metadata_file,
-        3, output_filename, version_number, False)
+        3, output_filename, version_number, False, storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.write_metadata_file,
-        root_signable, 3, version_number, False)
+        root_signable, 3, version_number, False, storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.write_metadata_file,
-        root_signable, output_filename, '3', False)
+        root_signable, output_filename, '3', False, storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_lib.write_metadata_file,
-        root_signable, output_filename, version_number, 3)
+        root_signable, output_filename, storage_backend, version_number, 3)
 
 
 
@@ -774,9 +720,11 @@ def test__generate_and_write_metadata(self):
     tuf.roledb.add_role('obsolete_role', targets_roleinfo,
         repository_name=repository_name)
 
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     repo_lib._generate_and_write_metadata('obsolete_role', obsolete_metadata,
-        targets_directory, metadata_directory, consistent_snapshot=False,
-        filenames=None, repository_name=repository_name)
+        targets_directory, metadata_directory, storage_backend,
+        consistent_snapshot=False, filenames=None,
+        repository_name=repository_name)
 
     snapshot_filepath = os.path.join('repository_data', 'repository',
                                      'metadata', 'snapshot.json')
@@ -843,7 +791,8 @@ def test__load_top_level_metadata(self):
     signable = securesystemslib.util.load_json_file(os.path.join(metadata_directory, 'root.json'))
     signable['signatures'].append(signable['signatures'][0])
 
-    repo_lib.write_metadata_file(signable, root_file, 8, False)
+    storage_backend = securesystemslib.storage.FilesystemBackend()
+    repo_lib.write_metadata_file(signable, root_file, 8, False, storage_backend)
 
     # Attempt to load a repository that contains a compressed Root file.
     repository = repo_tool.create_new_repository(repository_directory, repository_name)

tests/test_repository_tool.py

@@ -48,6 +48,7 @@
 import securesystemslib.exceptions
 
 import securesystemslib
+import securesystemslib.storage
 import six
 
 logger = logging.getLogger(__name__)
@@ -96,20 +97,22 @@ def tearDown(self):
   def test_init(self):
     # Test normal case.
     repository_name = 'test_repository'
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     repository = repo_tool.Repository('repository_directory/',
-        'metadata_directory/', 'targets_directory/', repository_name)
+        'metadata_directory/', 'targets_directory/', storage_backend,
+        repository_name)
     self.assertTrue(isinstance(repository.root, repo_tool.Root))
     self.assertTrue(isinstance(repository.snapshot, repo_tool.Snapshot))
     self.assertTrue(isinstance(repository.timestamp, repo_tool.Timestamp))
     self.assertTrue(isinstance(repository.targets, repo_tool.Targets))
 
     # Test improperly formatted arguments.
-    self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository, 3,
-                      'metadata_directory/', 'targets_directory')
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository,
-                      'repository_directory', 3, 'targets_directory')
+                      storage_backend, 3, 'metadata_directory/', 'targets_directory')
+    self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository,
+                      'repository_directory', storage_backend, 3, 'targets_directory')
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository,
-                      'repository_directory', 'metadata_directory', 3)
+                      'repository_directory', 'metadata_directory', 3, storage_backend)
 
 
 
@@ -1818,14 +1821,9 @@ def test_create_new_repository(self):
         repo_tool.create_new_repository, 3, repository_name)
 
     # For testing purposes, try to create a repository directory that
-    # fails due to a non-errno.EEXIST exception raised.  create_new_repository()
-    # should only pass for OSError (errno.EEXIST).
-    try:
-      repo_tool.create_new_repository('bad' * 2000, repository_name)
-
-    except OSError as e:
-      # errno.ENOENT is raised in Windows.
-      self.assertTrue(e.errno == errno.ENAMETOOLONG or e.errno == errno.ENOENT)
+    # fails due to a non-errno.EEXIST exception raised.
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+        repo_tool.create_new_repository, 'bad' * 2000, repository_name)
 
     # Reset the 'repository_directory' so that the metadata and targets
     # directories can be tested likewise.
@@ -1836,12 +1834,8 @@ def test_create_new_repository(self):
       tuf.repository_tool.METADATA_STAGED_DIRECTORY_NAME
     tuf.repository_tool.METADATA_STAGED_DIRECTORY_NAME = 'bad' * 2000
 
-    try:
-      repo_tool.create_new_repository(repository_directory, repository_name)
-
-    except OSError as e:
-      # errno.ENOENT is raised in Windows.
-      self.assertTrue(e.errno == errno.ENAMETOOLONG or e.errno == errno.ENOENT)
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+        repo_tool.create_new_repository, repository_directory, repository_name)
 
     # Reset metadata staged directory so that the targets directory can be
     # tested...
@@ -1851,12 +1845,8 @@ def test_create_new_repository(self):
     original_targets_directory = tuf.repository_tool.TARGETS_DIRECTORY_NAME
     tuf.repository_tool.TARGETS_DIRECTORY_NAME = 'bad' * 2000
 
-    try:
-      repo_tool.create_new_repository(repository_directory, repository_name)
-
-    except OSError as e:
-      # errno.ENOENT is raised in Windows.
-      self.assertTrue(e.errno == errno.ENAMETOOLONG or e.errno == errno.ENOENT)
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+         repo_tool.create_new_repository, repository_directory, repository_name)
 
     tuf.repository_tool.TARGETS_DIRECTORY_NAME = \
       original_targets_directory
@@ -1943,8 +1933,10 @@ def test_dump_signable_metadata(self):
     metadata_content = repo_tool.dump_signable_metadata(targets_metadata_file)
 
     # Test for an invalid targets metadata file..
-    self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.dump_signable_metadata, 1)
-    self.assertRaises(IOError, repo_tool.dump_signable_metadata, 'bad file path')
+    self.assertRaises(securesystemslib.exceptions.FormatError,
+        repo_tool.dump_signable_metadata, 1)
+    self.assertRaises(securesystemslib.exceptions.StorageError,
+        repo_tool.dump_signable_metadata, 'bad file path')
 
 
 

tests/test_root_versioning_integration.py

@@ -41,6 +41,7 @@
 import tuf.repository_tool as repo_tool
 
 import securesystemslib
+import securesystemslib.storage
 
 logger = logging.getLogger(__name__)
 
@@ -63,21 +64,23 @@ def tearDown(self):
 
   def test_init(self):
     # Test normal case.
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     repository = repo_tool.Repository('repository_directory/',
                                       'metadata_directory/',
-                                      'targets_directory/')
+                                      'targets_directory/',
+                                      storage_backend)
     self.assertTrue(isinstance(repository.root, repo_tool.Root))
     self.assertTrue(isinstance(repository.snapshot, repo_tool.Snapshot))
     self.assertTrue(isinstance(repository.timestamp, repo_tool.Timestamp))
     self.assertTrue(isinstance(repository.targets, repo_tool.Targets))
 
     # Test improperly formatted arguments.
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository, 3,
-                      'metadata_directory/', 'targets_directory')
+                      'metadata_directory/', 'targets_directory', storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository,
-                      'repository_directory', 3, 'targets_directory')
+                      'repository_directory', 3, 'targets_directory', storage_backend)
     self.assertRaises(securesystemslib.exceptions.FormatError, repo_tool.Repository,
-                      'repository_directory', 'metadata_directory', 3)
+                      'repository_directory', 'metadata_directory', storage_backend, 3)
 
 
 

tests/test_updater.py

@@ -431,6 +431,7 @@ def test_1__refresh_must_not_count_duplicate_keyids_towards_threshold(self):
     repository.root.threshold = 2
     repository.root.load_signing_key(self.role_keys['root']['private'])
 
+    storage_backend = securesystemslib.storage.FilesystemBackend()
     # The client uses the threshold from the previous root file to verify the
     # new root. Thus we need to make two updates so that the threshold used for
     # verification becomes 2. I.e. we bump the version, sign twice with the
@@ -450,7 +451,8 @@ def test_1__refresh_must_not_count_duplicate_keyids_towards_threshold(self):
       # catch the unmet threshold.
       # We also skip writing to 'metadata.staged' and copying to 'metadata' and
       # instead write directly to 'metadata'
-      repo_lib.write_metadata_file(signed_metadata, live_root_path, info["version"], True)
+      repo_lib.write_metadata_file(signed_metadata, live_root_path,
+          info["version"], True, storage_backend)
 
 
     # Update from current '1.root.json' to '3.root.json' on client and assert