Merge pull request #974 from lukpueh/fix-signature-threshold

Fix signature threshold

tests/test_sig.py

@@ -31,6 +31,7 @@
 
 import unittest
 import logging
+import copy
 
 import tuf
 import tuf.log
@@ -384,6 +385,68 @@ def test_verify_single_key(self):
     tuf.roledb.remove_role('Root')
 
 
+
+  def test_verify_must_not_count_duplicate_keyids_towards_threshold(self):
+    # Create and sign dummy metadata twice with same key
+    # Note that we use the non-deterministic rsassa-pss signing scheme, so
+    # creating the signature twice shows that we don't only detect duplicate
+    # signatures but also different signatures from the same key.
+    signable = {"signed" : "test", "signatures" : []}
+    signed = securesystemslib.formats.encode_canonical(
+        signable["signed"]).encode("utf-8")
+    signable["signatures"].append(
+        securesystemslib.keys.create_signature(KEYS[0], signed))
+    signable["signatures"].append(
+        securesystemslib.keys.create_signature(KEYS[0], signed))
+
+    # 'get_signature_status' uses keys from keydb for verification
+    tuf.keydb.add_key(KEYS[0])
+
+    # Assert that 'get_signature_status' returns two good signatures ...
+    status = tuf.sig.get_signature_status(
+        signable, "root", keyids=[KEYS[0]["keyid"]], threshold=2)
+    self.assertTrue(len(status["good_sigs"]) == 2)
+
+    # ... but only one counts towards the threshold
+    self.assertFalse(
+        tuf.sig.verify(signable, "root", keyids=[KEYS[0]["keyid"]], threshold=2))
+
+    # Clean-up keydb
+    tuf.keydb.remove_key(KEYS[0]["keyid"])
+
+
+
+  def test_verify_count_different_keyids_for_same_key_towards_threshold(self):
+    # Create and sign dummy metadata twice with same key but different keyids
+    signable = {"signed" : "test", "signatures" : []}
+    key_sha256 = copy.deepcopy(KEYS[0])
+    key_sha256["keyid"] = "deadbeef256"
+
+    key_sha512 = copy.deepcopy(KEYS[0])
+    key_sha512["keyid"] = "deadbeef512"
+
+    signed = securesystemslib.formats.encode_canonical(
+        signable["signed"]).encode("utf-8")
+    signable["signatures"].append(
+        securesystemslib.keys.create_signature(key_sha256, signed))
+    signable["signatures"].append(
+        securesystemslib.keys.create_signature(key_sha512, signed))
+
+    # 'get_signature_status' uses keys from keydb for verification
+    tuf.keydb.add_key(key_sha256)
+    tuf.keydb.add_key(key_sha512)
+
+    # Assert that both keys count towards threshold although its the same key
+    keyids = [key_sha256["keyid"], key_sha512["keyid"]]
+    self.assertTrue(
+        tuf.sig.verify(signable, "root", keyids=keyids, threshold=2))
+
+    # Clean-up keydb
+    tuf.keydb.remove_key(key_sha256["keyid"])
+    tuf.keydb.remove_key(key_sha512["keyid"])
+
+
+
   def test_verify_unrecognized_sig(self):
     signable = {'signed' : 'test', 'signatures' : []}
     signed = securesystemslib.formats.encode_canonical(signable['signed']).encode('utf-8')

tests/test_updater.py

@@ -425,6 +425,53 @@ def test_1__update_versioninfo(self):
 
 
 
+  def test_1__refresh_must_not_count_duplicate_keyids_towards_threshold(self):
+    # Update root threshold on the server repository and sign twice with 1 key
+    repository = repo_tool.load_repository(self.repository_directory)
+    repository.root.threshold = 2
+    repository.root.load_signing_key(self.role_keys['root']['private'])
+
+    # 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
+    # same key and write to disk '2.root.json' and '3.root.json'.
+    for version in [2, 3]:
+      repository.root.version = version
+      info = tuf.roledb.get_roleinfo("root")
+      metadata = repo_lib.generate_root_metadata(
+          info["version"], info["expires"], False)
+      signed_metadata = repo_lib.sign_metadata(
+          metadata, info["keyids"], "root.json", "default")
+      signed_metadata["signatures"].append(signed_metadata["signatures"][0])
+      live_root_path = os.path.join(
+          self.repository_directory, "metadata", "root.json")
+
+      # Bypass server side verification in 'write' or 'writeall', which would
+      # 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)
+
+
+    # Update from current '1.root.json' to '3.root.json' on client and assert
+    # raise of 'BadSignatureError' (caused by unmet signature threshold).
+    try:
+      self.repository_updater.refresh()
+
+    except tuf.exceptions.NoWorkingMirrorError as e:
+      mirror_errors = list(e.mirror_errors.values())
+      self.assertTrue(len(mirror_errors) == 1)
+      self.assertTrue(
+          isinstance(mirror_errors[0],
+          securesystemslib.exceptions.BadSignatureError))
+      self.assertEqual(
+          str(mirror_errors[0]),
+          repr("root") + " metadata has bad signature.")
+
+    else:
+      self.fail(
+          "Expected a NoWorkingMirrorError composed of one BadSignatureError")
+
 
   def test_1__update_fileinfo(self):
       # Tests

tuf/sig.py

@@ -71,11 +71,23 @@ 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 of these signatures can be determined.  This method will iterate
-    the signatures in 'signable' and enumerate all the keys that are valid,
-    invalid, unrecognized, or unauthorized.
+    'signable'. Signatures in the returned dictionary are identified by the
+    signature keyid and can have a status of either:
+
+    * bad -- Invalid signature
+    * good -- Valid signature from key that is available in 'tuf.keydb', and is
+      authorized for the passed role as per 'tuf.roledb' (authorization may be
+      overwritten by passed 'keyids').
+    * unknown -- Signature from key that is not available in 'tuf.keydb', or if
+      'role' is None.
+    * unknown signing schemes -- Signature from key with unknown signing
+      scheme.
+    * untrusted -- Valid signature from key that is available in 'tuf.keydb',
+      but is not trusted for the passed role as per 'tuf.roledb' or the passed
+      'keyids'.
+
+    NOTE: The result may contain duplicate keyids or keyids that reference the
+    same key, if 'signable' lists multiple signatures from the same key.
 
   <Arguments>
     signable:
@@ -87,7 +99,7 @@ def get_signature_status(signable, role=None, repository_name='default',
       Conformant to tuf.formats.SIGNABLE_SCHEMA.
 
     role:
-      TUF role (e.g., 'root', 'targets', 'snapshot').
+      TUF role string (e.g. 'root', 'targets', 'snapshot' or timestamp).
 
     threshold:
       Rather than reference the role's threshold as set in tuf.roledb.py, use
@@ -133,22 +145,6 @@ def get_signature_status(signable, role=None, repository_name='default',
 
   # The signature status dictionary returned.
   signature_status = {}
-
-  # The fields of the signature_status dict, where each field stores keyids.  A
-  # description of each field:
-  #
-  # good_sigs = keys confirmed to have produced 'sig' using 'signed', which are
-  # associated with 'role';
-  #
-  # bad_sigs = negation of good_sigs;
-  #
-  # unknown_sigs = keys not found in the 'keydb' database;
-  #
-  # untrusted_sigs = keys that are not in the list of keyids associated with
-  # 'role';
-  #
-  # unknown_signing_scheme = signing schemes specified in keys that are
-  # unsupported;
   good_sigs = []
   bad_sigs = []
   unknown_sigs = []
@@ -240,18 +236,26 @@ def verify(signable, role, repository_name='default', threshold=None,
     keyids=None):
   """
   <Purpose>
-    Verify whether the authorized signatures of 'signable' meet the minimum
-    required by 'role'.  Authorized signatures are those with valid keys
-    associated with 'role'.  'signable' must conform to SIGNABLE_SCHEMA
-    and 'role' must not equal 'None' or be less than zero.
+    Verify that 'signable' has a valid threshold of authorized signatures
+    identified by unique keyids. The threshold and whether a keyid is
+    authorized is determined by querying the 'threshold' and 'keyids' info for
+    the passed 'role' in 'tuf.roledb'. Both values can be overwritten by
+    passing the 'threshold' or 'keyids' arguments.
+
+    NOTE:
+    - Signatures with identical authorized keyids only count towards the
+      threshold once.
+    - Signatures with different authorized keyids each count towards the
+      threshold, even if the keyids identify the same key.
 
   <Arguments>
     signable:
-      A dictionary containing a list of signatures and a 'signed' identifier.
+      A dictionary containing a list of signatures and a 'signed' identifier
+      that conforms to SIGNABLE_SCHEMA, e.g.:
       signable = {'signed':, 'signatures': [{'keyid':, 'method':, 'sig':}]}
 
     role:
-      TUF role (e.g., 'root', 'targets', 'snapshot').
+      TUF role string (e.g. 'root', 'targets', 'snapshot' or timestamp).
 
     threshold:
       Rather than reference the role's threshold as set in tuf.roledb.py, use
@@ -278,8 +282,8 @@ def verify(signable, role, repository_name='default', threshold=None,
     get_signature_status() will be caught here and re-raised.
 
   <Returns>
-    Boolean.  True if the number of good signatures >= the role's threshold,
-    False otherwise.
+    Boolean.  True if the number of good unique (by keyid) signatures >= the
+    role's threshold, False otherwise.
   """
 
   tuf.formats.SIGNABLE_SCHEMA.check_match(signable)
@@ -303,7 +307,7 @@ def verify(signable, role, repository_name='default', threshold=None,
   if threshold is None or threshold <= 0: #pragma: no cover
     raise securesystemslib.exceptions.Error("Invalid threshold: " + repr(threshold))
 
-  return len(good_sigs) >= threshold
+  return len(set(good_sigs)) >= threshold