diff --git a/tests/test_repository_lib.py b/tests/test_repository_lib.py index ff30d725..36c2d4bb 100755 --- a/tests/test_repository_lib.py +++ b/tests/test_repository_lib.py @@ -793,6 +793,15 @@ def test_write_metadata_file(self): version_number, compression_algorithms, 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.conf.CONSISTENT_METHOD = 'hard_link' + os.remove(output_filename) + repo_lib.write_metadata_file(root_signable, output_filename, + version_number, + compression_algorithms, + consistent_snapshot=True) # Reset CONSISTENT_METHOD so that subsequent tests work as expected. tuf.conf.CONSISTENT_METHOD = 'copy' diff --git a/tests/test_repository_tool.py b/tests/test_repository_tool.py index 30938a6f..95dfd85e 100755 --- a/tests/test_repository_tool.py +++ b/tests/test_repository_tool.py @@ -347,6 +347,19 @@ def test_writeall(self): # successfully. repo_tool.load_repository(repository_directory) + # Verify the behavior of marking and unmarking roles as dirty. + # We begin by ensuring that writeall() cleared the list of dirty roles.. + self.assertEqual([], tuf.roledb.get_dirty_roles()) + + repository.mark_dirty(['root', 'timestamp']) + self.assertEqual(['root', 'timestamp'], sorted(tuf.roledb.get_dirty_roles())) + repository.unmark_dirty(['root']) + self.assertEqual(['timestamp'], tuf.roledb.get_dirty_roles()) + + # Ensure status() does not leave behind any dirty roles. + repository.status() + self.assertEqual(['timestamp'], tuf.roledb.get_dirty_roles()) + # Test improperly formatted arguments. self.assertRaises(tuf.FormatError, repository.writeall, 3, False) self.assertRaises(tuf.FormatError, repository.writeall, False, 3) diff --git a/tests/test_roledb.py b/tests/test_roledb.py index 6149dd5d..13713979 100755 --- a/tests/test_roledb.py +++ b/tests/test_roledb.py @@ -701,6 +701,36 @@ def test_mark_dirty(self): self.assertRaises(tuf.InvalidNameError, tuf.roledb.mark_dirty, ['dirty_role'], 'non-existent') + + + def test_unmark_dirty(self): + # Add a dirty role to roledb. + rolename = 'targets' + roleinfo1 = {'keyids': ['123'], 'threshold': 1} + tuf.roledb.add_role(rolename, roleinfo1) + rolename2 = 'dirty_role' + roleinfo2 = {'keyids': ['123'], 'threshold': 2} + tuf.roledb.add_role(rolename2, roleinfo2) + mark_role_as_dirty = True + tuf.roledb.update_roleinfo(rolename, roleinfo1, mark_role_as_dirty) + # Note: The 'default' repository is searched if the repository name is + # not given to get_dirty_roles(). + self.assertEqual([rolename], tuf.roledb.get_dirty_roles()) + tuf.roledb.update_roleinfo(rolename2, roleinfo2, mark_role_as_dirty) + + tuf.roledb.unmark_dirty(['dirty_role']) + self.assertEqual([rolename], sorted(tuf.roledb.get_dirty_roles())) + tuf.roledb.unmark_dirty(['targets']) + self.assertEqual([], sorted(tuf.roledb.get_dirty_roles())) + + # What happens for a role that isn't dirty? unmark_dirty() should just + # log a message. + tuf.roledb.unmark_dirty(['unknown_role']) + + # Verify that a role cannot be unmarked as dirty for a non-existent + # repository. + self.assertRaises(tuf.InvalidNameError, tuf.roledb.unmark_dirty, + ['dirty_role'], 'non-existent') def _test_rolename(self, test_function): diff --git a/tuf/repository_lib.py b/tuf/repository_lib.py index e03da42a..76914642 100755 --- a/tuf/repository_lib.py +++ b/tuf/repository_lib.py @@ -1917,7 +1917,7 @@ def sign_metadata(metadata_object, keyids, filename): logger.warning('Unable to create signature for keyid: ' + repr(keyid)) else: - logger.warning('Private key unset. Skipping: ' + repr(keyid)) + logger.debug('Private key unset. Skipping: ' + repr(keyid)) else: raise tuf.Error('The keydb contains a key with an invalid key type.') @@ -2039,15 +2039,18 @@ def write_metadata_file(metadata, filename, version_number, # to its expected filename (e.g., root.json). The option of either # creating a copy or link should be configurable in tuf.conf.py. if (tuf.conf.CONSISTENT_METHOD == 'copy'): - logger.info('Pointing ' + repr(filename) + ' to the consistent snapshot.') + logger.debug('Pointing ' + repr(filename) + ' to the consistent snapshot.') shutil.copyfile(written_consistent_filename, written_filename) elif (tuf.conf.CONSISTENT_METHOD == 'hard_link'): logger.info('Hard linking ' + repr(written_consistent_filename)) - # 'written_filename' must not exist, otherwise os.link complains. + # 'written_filename' must not exist, otherwise os.link() complains. if os.path.exists(written_filename): os.remove(written_filename) + + else: + logger.debug(repr(written_filename) + ' does not exist.') os.link(written_consistent_filename, written_filename) @@ -2150,7 +2153,7 @@ def _write_compressed_metadata(file_object, compressed_filename, # Move the 'tuf.util.TempFile' object to one of the filenames so that it is # saved and the temporary file closed. if not os.path.exists(consistent_filename): - logger.info('Saving ' + repr(consistent_filename)) + logger.debug('Saving ' + repr(consistent_filename)) file_object.move(consistent_filename) else: @@ -2201,6 +2204,16 @@ def _log_status_of_top_level_roles(targets_directory, metadata_directory): # Do the top-level roles contain a valid threshold of signatures? Top-level # metadata is verified in Root -> Targets -> Snapshot -> Timestamp order. # Verify the metadata of the Root role. + dirty_rolenames = tuf.roledb.get_dirty_roles() + + root_roleinfo = tuf.roledb.get_roleinfo('root') + root_is_dirty = None + if 'root' in dirty_rolenames: + root_is_dirty = True + + else: + root_is_dirty = False + try: signable, root_filename = \ _generate_and_write_metadata('root', root_filename, @@ -2213,7 +2226,20 @@ def _log_status_of_top_level_roles(targets_directory, metadata_directory): _log_status('root', e.signable) return + finally: + tuf.roledb.unmark_dirty(['root']) + tuf.roledb.update_roleinfo('root', root_roleinfo, + mark_role_as_dirty=root_is_dirty) + # Verify the metadata of the Targets role. + targets_roleinfo = tuf.roledb.get_roleinfo('targets') + targets_is_dirty = None + if 'targets' in dirty_rolenames: + targets_is_dirty = True + + else: + targets_is_dirty = False + try: signable, targets_filename = \ _generate_and_write_metadata('targets', targets_filename, @@ -2223,8 +2249,21 @@ def _log_status_of_top_level_roles(targets_directory, metadata_directory): except tuf.UnsignedMetadataError as e: _log_status('targets', e.signable) return + + finally: + tuf.roledb.unmark_dirty(['targets']) + tuf.roledb.update_roleinfo('targets', targets_roleinfo, + mark_role_as_dirty=targets_is_dirty) # Verify the metadata of the snapshot role. + snapshot_roleinfo = tuf.roledb.get_roleinfo('snapshot') + snapshot_is_dirty = None + if 'snapshot' in dirty_rolenames: + snapshot_is_dirty = True + + else: + snapshot_is_dirty = False + filenames = {'root': root_filename, 'targets': targets_filename} try: signable, snapshot_filename = \ @@ -2237,7 +2276,20 @@ def _log_status_of_top_level_roles(targets_directory, metadata_directory): _log_status('snapshot', e.signable) return + finally: + tuf.roledb.unmark_dirty(['snapshot']) + tuf.roledb.update_roleinfo('snapshot', snapshot_roleinfo, + mark_role_as_dirty=snapshot_is_dirty) + # Verify the metadata of the Timestamp role. + timestamp_roleinfo = tuf.roledb.get_roleinfo('timestamp') + timestamp_is_dirty = None + if 'timestamp' in dirty_rolenames: + timestamp_is_dirty = True + + else: + timestamp_is_dirty = False + filenames = {'snapshot': snapshot_filename} try: signable, timestamp_filename = \ @@ -2249,8 +2301,12 @@ def _log_status_of_top_level_roles(targets_directory, metadata_directory): except tuf.UnsignedMetadataError as e: _log_status('timestamp', e.signable) return - - + + finally: + tuf.roledb.unmark_dirty(['timestamp']) + tuf.roledb.update_roleinfo('timestamp', timestamp_roleinfo, + mark_role_as_dirty=timestamp_is_dirty) + def _log_status(rolename, signable): diff --git a/tuf/repository_tool.py b/tuf/repository_tool.py index 5543199d..cdf77b5e 100755 --- a/tuf/repository_tool.py +++ b/tuf/repository_tool.py @@ -279,7 +279,9 @@ def writeall(self, consistent_snapshot=False, compression_algorithms=['gz']): self._targets_directory, self._metadata_directory, consistent_snapshot, filenames) - + + tuf.roledb.unmark_dirty(dirty_rolenames) + # Delete the metadata of roles no longer in 'tuf.roledb'. Obsolete roles # may have been revoked and should no longer have their metadata files # available on disk, otherwise loading a repository may unintentionally @@ -339,6 +341,9 @@ def write(self, rolename, consistent_snapshot=False, increment_version_number=Tr filenames=filenames, allow_partially_signed=True, increment_version_number=increment_version_number) + + # Ensure 'rolename' is no longer marked as dirty after the successful write(). + tuf.roledb.unmark_dirty([rolename]) @@ -439,6 +444,30 @@ def mark_dirty(self, roles): """ tuf.roledb.mark_dirty(roles) + + + + def unmark_dirty(self, roles): + """ + + No longer mark the list of 'roles' as dirty. + + + roles: + A list of roles to mark as dirty. on the next write, these roles + will be written to disk. + + + None. + + + None. + + + None. + """ + + tuf.roledb.unmark_dirty(roles) diff --git a/tuf/roledb.py b/tuf/roledb.py index 2fe36df0..a4338213 100755 --- a/tuf/roledb.py +++ b/tuf/roledb.py @@ -471,6 +471,51 @@ def mark_dirty(roles, repository_name='default'): +def unmark_dirty(roles, repository_name='default'): + """ + + No longer mark the roles in 'roles' as dirty. + + + repository_name: + The name of the repository to get the dirty roles. If not supplied, the + 'default' repository is searched. + + roles: + A list of roles that should no longer be marked as dirty. + + + tuf.FormatError, if the arguments are improperly formatted. + + tuf.InvalidNameError, if 'repository_name' does not exist in the role + database. + + + None. + + + None. + """ + + # Are the arguments properly formatted? If not, raise tuf.FormatError. + tuf.formats.NAMES_SCHEMA.check_match(roles) + tuf.formats.NAME_SCHEMA.check_match(repository_name) + + global _roledb_dict + global _dirty_roles + + if repository_name not in _roledb_dict or repository_name not in _dirty_roles: + raise tuf.InvalidNameError('Repository name does not exist: ' + repository_name) + + for role in roles: + try: + _dirty_roles[repository_name].remove(role) + + except (KeyError, ValueError): + logger.debug(repr(role) + ' is not dirty.') + + + def role_exists(rolename, repository_name='default'): """