Merge pull request #389 from vladimir-v-diaz/develop

Fix for repository.status()

Author: vladimir-v-diaz

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'

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)

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):

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):

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):
+    """
+    <Purpose>
+      No longer mark the list of 'roles' as dirty.
+
+    <Arguments>
+      roles:
+        A list of roles to mark as dirty.  on the next write, these roles
+        will be written to disk.
+
+    <Exceptions>
+      None.
+    
+    <Side Effects>
+      None.
+
+    <Returns>
+      None.
+    """
+  
+    tuf.roledb.unmark_dirty(roles)
 
 
 

tuf/roledb.py

@@ -471,6 +471,51 @@ def mark_dirty(roles, repository_name='default'):
 
 
 
+def unmark_dirty(roles, repository_name='default'):
+  """
+  <Purpose>
+    No longer mark the roles in 'roles' as dirty.
+
+  <Arguments>
+    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.
+
+  <Exceptions>
+    tuf.FormatError, if the arguments are improperly formatted.
+
+    tuf.InvalidNameError, if 'repository_name' does not exist in the role
+    database.
+
+  <Side Effects>
+    None.
+
+  <Returns>
+    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'):
   """
   <Purpose>