From 03a1caa1a8d9fa6df9f7b59af074c78bea1d4919 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Thu, 1 Feb 2024 15:27:45 +0200 Subject: [PATCH 01/13] Metadata API: Refactor VerificationResult This is an API break as VerificationResult changes: * Now contains threshold * Now contains Keys and not just keyids Note that there is a small edge case functionality change: * if the role does not have a key for the keyid, then we no longer include that key in "unsigned" I think that is an acceptable change. Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 42 +++++++++++++++--------------------------- 1 file changed, 15 insertions(+), 27 deletions(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 205678a3d1..a7d08da6cd 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -44,7 +44,6 @@ Iterator, List, Optional, - Set, Tuple, Type, TypeVar, @@ -654,29 +653,21 @@ class VerificationResult: Attributes: verified: True, if threshold of signatures is met. - signed: Set of delegated keyids, which validly signed. - unsigned: Set of delegated keyids, which did not validly sign. - + threshold: Number of required signatures. + signed: dict of keyid:Key containing the keys that have signed. + unsigned: dict of keyid:Key containing the keys that have not signed. """ - verified: bool - signed: Set[str] - unsigned: Set[str] + threshold: int + signed: Dict[str, Key] + unsigned: Dict[str, Key] def __bool__(self) -> bool: return self.verified - def union(self, other: "VerificationResult") -> "VerificationResult": - """Combine two verification results. - - Can be used to verify, if root metadata is signed by the threshold of - keys of previous root and the threshold of keys of itself. - """ - return VerificationResult( - self.verified and other.verified, - self.signed | other.signed, - self.unsigned | other.unsigned, - ) + @property + def verified(self) -> bool: + return len(self.signed) >= self.threshold class _DelegatorMixin(metaclass=abc.ABCMeta): @@ -719,33 +710,30 @@ def get_verification_result( """ role = self.get_delegated_role(delegated_role) - signed = set() - unsigned = set() + signed = {} + unsigned = {} for keyid in role.keyids: try: key = self.get_key(keyid) except ValueError: - unsigned.add(keyid) logger.info("No key for keyid %s", keyid) continue if keyid not in signatures: - unsigned.add(keyid) + unsigned[keyid] = key logger.info("No signature for keyid %s", keyid) continue sig = signatures[keyid] try: key.verify_signature(sig, payload) - signed.add(keyid) + signed[keyid] = key except sslib_exceptions.UnverifiedSignatureError: - unsigned.add(keyid) + unsigned[keyid] = key logger.info("Key %s failed to verify %s", keyid, delegated_role) - return VerificationResult( - len(signed) >= role.threshold, signed, unsigned - ) + return VerificationResult(role.threshold, signed, unsigned) def verify_delegate( self, From 368bee82283ff180936d7e0f861c318797d428d7 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Thu, 1 Feb 2024 16:07:52 +0200 Subject: [PATCH 02/13] Metadata API: Implement RootVerificationResult This is a thin wrapper over two VerificationResults: useful when verifying root signatures. Now the API for getting verification results for root and the API for getting the results for other metadata is different. Client use cases can continue using verify_delegate() so should not be affected. Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 69 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 68 insertions(+), 1 deletion(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index a7d08da6cd..79f8bcbf0a 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -652,7 +652,6 @@ class VerificationResult: """Signature verification result for delegated role metadata. Attributes: - verified: True, if threshold of signatures is met. threshold: Number of required signatures. signed: dict of keyid:Key containing the keys that have signed. unsigned: dict of keyid:Key containing the keys that have not signed. @@ -667,9 +666,49 @@ def __bool__(self) -> bool: @property def verified(self) -> bool: + """True if threshold of signatures is met.""" return len(self.signed) >= self.threshold +@dataclass +class RootVerificationResult: + """Signature verification result for root metadata. + + Root must be verified by itself and the previous root version. This + dataclass represents that combination. For the edge case of first version + of root, the same VerificationResult can be used twice. + + Note that `signed` and `unsigned` correctness requires the underlying + VerificationResult keys to not conflict (no reusing the same keyid for + different keys). + + Attributes: + first: First underlying VerificationResult + second: Second underlying VerificationResult + """ + + first: VerificationResult + second: VerificationResult + + def __bool__(self) -> bool: + return self.verified + + @property + def verified(self) -> bool: + """True if threshold of signatures is met in both underlying VerificationResults.""" + return self.first.verified and self.second.verified + + @property + def signed(self) -> Dict[str, Key]: + """Dictionary of all signing keys that have signed, from both VerificationResults""" + return self.first.signed | self.second.signed + + @property + def unsigned(self) -> Dict[str, Key]: + """Dictionary of all signing keys that have not signed, from both VerificationResults""" + return self.first.unsigned | self.second.unsigned + + class _DelegatorMixin(metaclass=abc.ABCMeta): """Class that implements verify_delegate() for Root and Targets""" @@ -923,6 +962,34 @@ def get_key(self, keyid: str) -> Key: # noqa: D102 return self.keys[keyid] + def get_root_verification_result( + self, + other: "Root", + payload: bytes, + signatures: Dict[str, Signature], + ) -> RootVerificationResult: + """Return signature threshold verification result for two root roles. + + Verify root metadata with two roles (the root role from `self` and + `other`). If you have only one role (in the case of root v1) you can + provide the same Root as both `self` and `other`. + + NOTE: Unlike `verify_delegate()` this method does not raise, if the + root metadata is not fully verified. + + Args: + other: The other `Root` to verify payload with + payload: Signed payload bytes for root + signatures: Signatures over payload bytes + + Raises: + ValueError: no delegation was found for ``delegated_role``. + """ + return RootVerificationResult( + self.get_verification_result(Root.type, payload, signatures), + other.get_verification_result(Root.type, payload, signatures), + ) + class BaseFile: """A base class of ``MetaFile`` and ``TargetFile``. From 506b40d93d285e1fc497df21a6779786a3d7e87d Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Thu, 1 Feb 2024 19:54:39 +0200 Subject: [PATCH 03/13] tests: Update to new VerificationResult Changes are * expected result changes (like the handling of keyids without keys) * test refactoring to have access to the Key * Removal of union test * use the fact that VerificationResult is Truthy in asserts (to get 1 more line of coverage) Signed-off-by: Jussi Kukkonen --- tests/test_api.py | 84 +++++++++++++++++------------------------------ 1 file changed, 30 insertions(+), 54 deletions(-) diff --git a/tests/test_api.py b/tests/test_api.py index 517ff5bdf8..8812647868 100755 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -47,7 +47,6 @@ TargetFile, Targets, Timestamp, - VerificationResult, ) from tuf.api.serialization import DeserializationError, SerializationError from tuf.api.serialization.json import JSONSerializer @@ -475,92 +474,69 @@ def test_signed_get_verification_result(self) -> None: # Setup: Load test metadata and keys root_path = os.path.join(self.repo_dir, "metadata", "root.json") root = Metadata[Root].from_file(root_path) - initial_root_keyids = root.signed.roles[Root.type].keyids - self.assertEqual(len(initial_root_keyids), 1) - key1_id = initial_root_keyids[0] - key2 = self.keystore[Timestamp.type] - key2_id = key2["keyid"] + + key1_id = root.signed.roles[Root.type].keyids[0] + key1 = root.signed.get_key(key1_id) + + key2_id = root.signed.roles[Timestamp.type].keyids[0] + key2 = root.signed.get_key(key2_id) + priv_key2 = self.keystore[Timestamp.type] + key3_id = "123456789abcdefg" - key4 = self.keystore[Snapshot.type] - key4_id = key4["keyid"] + priv_key4 = self.keystore[Snapshot.type] + key4_id = priv_key4["keyid"] # Test: 1 authorized key, 1 valid signature result = root.signed.get_verification_result( Root.type, root.signed_bytes, root.signatures ) - self.assertTrue(result.verified) - self.assertEqual(result.signed, {key1_id}) - self.assertEqual(result.unsigned, set()) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1}) + self.assertEqual(result.unsigned, {}) # Test: 2 authorized keys, 1 invalid signature # Adding a key, i.e. metadata change, invalidates existing signature - root.signed.add_key( - SSlibKey.from_securesystemslib_key(key2), - Root.type, - ) + root.signed.add_key(key2, Root.type) result = root.signed.get_verification_result( Root.type, root.signed_bytes, root.signatures ) - self.assertFalse(result.verified) - self.assertEqual(result.signed, set()) - self.assertEqual(result.unsigned, {key1_id, key2_id}) + self.assertFalse(result) + self.assertEqual(result.signed, {}) + self.assertEqual(result.unsigned, {key1_id: key1, key2_id: key2}) # Test: 3 authorized keys, 1 invalid signature, 1 key missing key data - # Adding a keyid w/o key, fails verification the same as no signature - # or an invalid signature for that key + # Adding a keyid w/o key, fails verification but this key is not listed + # in unsigned root.signed.roles[Root.type].keyids.append(key3_id) result = root.signed.get_verification_result( Root.type, root.signed_bytes, root.signatures ) - self.assertFalse(result.verified) - self.assertEqual(result.signed, set()) - self.assertEqual(result.unsigned, {key1_id, key2_id, key3_id}) + self.assertFalse(result) + self.assertEqual(result.signed, {}) + self.assertEqual(result.unsigned, {key1_id: key1, key2_id: key2}) # Test: 3 authorized keys, 1 valid signature, 1 invalid signature, 1 # key missing key data - root.sign(SSlibSigner(key2), append=True) + root.sign(SSlibSigner(priv_key2), append=True) result = root.signed.get_verification_result( Root.type, root.signed_bytes, root.signatures ) - self.assertTrue(result.verified) - self.assertEqual(result.signed, {key2_id}) - self.assertEqual(result.unsigned, {key1_id, key3_id}) + self.assertTrue(result) + self.assertEqual(result.signed, {key2_id: key2}) + self.assertEqual(result.unsigned, {key1_id: key1}) # Test: 3 authorized keys, 1 valid signature, 1 invalid signature, 1 # key missing key data, 1 ignored unrelated signature - root.sign(SSlibSigner(key4), append=True) + root.sign(SSlibSigner(priv_key4), append=True) self.assertEqual( set(root.signatures.keys()), {key1_id, key2_id, key4_id} ) - self.assertTrue(result.verified) - self.assertEqual(result.signed, {key2_id}) - self.assertEqual(result.unsigned, {key1_id, key3_id}) + self.assertTrue(result) + self.assertEqual(result.signed, {key2_id: key2}) + self.assertEqual(result.unsigned, {key1_id: key1}) # See test_signed_verify_delegate for more related tests ... - def test_signed_verification_result_union(self) -> None: - # Test all possible "unions" (AND) of "verified" field - data = [ - (True, True, True), - (True, False, False), - (False, True, False), - (False, False, False), - ] - - for a_part, b_part, ab_part in data: - self.assertEqual( - VerificationResult(a_part, set(), set()).union( - VerificationResult(b_part, set(), set()) - ), - VerificationResult(ab_part, set(), set()), - ) - - # Test exemplary union (|) of "signed" and "unsigned" fields - a = VerificationResult(True, {"1"}, {"2"}) - b = VerificationResult(True, {"3"}, {"4"}) - ab = VerificationResult(True, {"1", "3"}, {"2", "4"}) - self.assertEqual(a.union(b), ab) - def test_key_class(self) -> None: # Test if from_securesystemslib_key removes the private key from keyval # of a securesystemslib key dictionary. From cd0fd5c2ffabb01302fba431c2eee8058a831807 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Thu, 1 Feb 2024 20:32:07 +0200 Subject: [PATCH 04/13] tests: Add tests for root verification This does much the same tests as test_signed_get_verification_result() above it does, just using two root roles. Signed-off-by: Jussi Kukkonen --- tests/test_api.py | 81 ++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 80 insertions(+), 1 deletion(-) diff --git a/tests/test_api.py b/tests/test_api.py index 8812647868..ad3da15928 100755 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -13,7 +13,7 @@ import sys import tempfile import unittest -from copy import copy +from copy import copy, deepcopy from datetime import datetime, timedelta from typing import Any, ClassVar, Dict, Optional @@ -537,6 +537,85 @@ def test_signed_get_verification_result(self) -> None: # See test_signed_verify_delegate for more related tests ... + def test_root_get_root_verification_result(self) -> None: + # Setup: Load test metadata and keys + root_path = os.path.join(self.repo_dir, "metadata", "root.json") + root = Metadata[Root].from_file(root_path) + + key1_id = root.signed.roles[Root.type].keyids[0] + key1 = root.signed.get_key(key1_id) + + key2_id = root.signed.roles[Timestamp.type].keyids[0] + key2 = root.signed.get_key(key2_id) + priv_key2 = self.keystore[Timestamp.type] + + priv_key4 = self.keystore[Snapshot.type] + + # other_root is only used as the other verifying role + other_root: Metadata[Root] = deepcopy(root) + + # Test: Verify with two roles that are the same + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1}) + self.assertEqual(result.unsigned, {}) + + # Test: Add a signer to other root (threshold still 1) + other_root.signed.add_key(key2, Root.type) + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1}) + self.assertEqual(result.unsigned, {key2_id: key2}) + + # Test: Increase threshold in other root + other_root.signed.roles[Root.type].threshold += 1 + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertFalse(result) + self.assertEqual(result.signed, {key1_id: key1}) + self.assertEqual(result.unsigned, {key2_id: key2}) + + # Test: Sign root with both keys + root.sign(SSlibSigner(priv_key2), append=True) + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) + self.assertEqual(result.unsigned, {}) + + # Test: Sign root with an unrelated key + root.sign(SSlibSigner(priv_key4), append=True) + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) + self.assertEqual(result.unsigned, {}) + + # Test: Remove key1 from other root + other_root.signed.revoke_key(key1_id, Root.type) + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertFalse(result) + self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) + self.assertEqual(result.unsigned, {}) + + # Test: Lower threshold in other root + other_root.signed.roles[Root.type].threshold -= 1 + result = root.signed.get_root_verification_result( + other_root.signed, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) + self.assertEqual(result.unsigned, {}) + def test_key_class(self) -> None: # Test if from_securesystemslib_key removes the private key from keyval # of a securesystemslib key dictionary. From dc11afc62ed349e574411c1b5fb8dba27ff88bfe Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Fri, 2 Feb 2024 11:02:27 +0200 Subject: [PATCH 05/13] Metadata API: Workaround for Python <3.9 dict unions are only supported in 3.9. Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 79f8bcbf0a..6a5cf222e8 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -701,12 +701,14 @@ def verified(self) -> bool: @property def signed(self) -> Dict[str, Key]: """Dictionary of all signing keys that have signed, from both VerificationResults""" - return self.first.signed | self.second.signed + # return a union of all signed (in python<3.9 this requires dict unpacking) + return {**self.first.signed, **self.second.signed} @property def unsigned(self) -> Dict[str, Key]: """Dictionary of all signing keys that have not signed, from both VerificationResults""" - return self.first.unsigned | self.second.unsigned + # return a union of all unsigned (in python<3.9 this requires dict unpacking) + return {**self.first.unsigned, **self.second.unsigned} class _DelegatorMixin(metaclass=abc.ABCMeta): From 26bdbbe20cdde8bab8ff3ff66c25c3f0b168079f Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Fri, 2 Feb 2024 11:04:01 +0200 Subject: [PATCH 06/13] Metadata API: Simplify verify_delegate() Now that VerificationResult has threshold, this can be simpler. Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 6a5cf222e8..23bbce64cd 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -802,10 +802,9 @@ def verify_delegate( delegated_role, payload, signatures ) if not result: - role = self.get_delegated_role(delegated_role) raise UnsignedMetadataError( f"{delegated_role} was signed by {len(result.signed)}/" - f"{role.threshold} keys" + f"{result.threshold} keys" ) From b8dbe307dbcbf1b3c927e536664d0c2fbe49d1ff Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Sat, 3 Feb 2024 11:13:49 +0200 Subject: [PATCH 07/13] examples: Use verification results in repo example This is an example of using the verification resutls in a repository. The only remaining tricky part is in _get_verification_result(): * has to figure out the delegating metadata (something we currently cannot provide in repository.Repository for the general case) * Needs a special case for first root Signed-off-by: Jussi Kukkonen --- examples/repository/_simplerepo.py | 51 +++++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 8 deletions(-) diff --git a/examples/repository/_simplerepo.py b/examples/repository/_simplerepo.py index ece4d99f59..668914622e 100644 --- a/examples/repository/_simplerepo.py +++ b/examples/repository/_simplerepo.py @@ -8,7 +8,7 @@ import logging from collections import defaultdict from datetime import datetime, timedelta -from typing import Dict, List +from typing import Dict, List, Union from securesystemslib import keys from securesystemslib.signer import Key, Signer, SSlibKey, SSlibSigner @@ -20,10 +20,13 @@ Metadata, MetaFile, Root, + RootVerificationResult, + Signed, Snapshot, TargetFile, Targets, Timestamp, + VerificationResult, ) from tuf.repository import Repository @@ -89,6 +92,27 @@ def targets_infos(self) -> Dict[str, MetaFile]: def snapshot_info(self) -> MetaFile: return self._snapshot_info + def _get_verification_result( + self, role: str, md: Metadata + ) -> Union[VerificationResult, RootVerificationResult]: + """Verify roles metadata using the existing repository metadata""" + if role == Root.type: + assert isinstance(md.signed, Root) + root = self.root() + if root.version == 0: + # special case first root + root = md.signed + return md.signed.get_root_verification_result( + root, md.signed_bytes, md.signatures + ) + if role in [Timestamp.type, Snapshot.type, Targets.type]: + delegator: Signed = self.root() + else: + delegator = self.targets() + return delegator.get_verification_result( + role, md.signed_bytes, md.signatures + ) + def open(self, role: str) -> Metadata: """Return current Metadata for role from 'storage' (or create a new one)""" @@ -112,6 +136,14 @@ def close(self, role: str, md: Metadata) -> None: for signer in self.signer_cache[role]: md.sign(signer, append=True) + # Double check that we only write verified metadata + vr = self._get_verification_result(role, md) + if not vr: + raise ValueError(f"Role {role} failed to verify") + keyids = [keyid[:7] for keyid in vr.signed] + verify_str = f"verified with keys [{', '.join(keyids)}]" + logger.debug("Role %s v%d: %s", role, md.signed.version, verify_str) + # store new metadata version, update version caches self.role_cache[role].append(md) if role == "snapshot": @@ -130,8 +162,6 @@ def add_target(self, path: str, content: str) -> None: with self.edit_targets() as targets: targets.targets[path] = TargetFile.from_data(path, data) - logger.debug("Targets v%d", targets.version) - # update snapshot, timestamp self.do_snapshot() self.do_timestamp() @@ -157,8 +187,6 @@ def submit_delegation(self, rolename: str, data: bytes) -> bool: logger.info("Failed to add delegation for %s: %s", rolename, e) return False - logger.debug("Targets v%d", targets.version) - # update snapshot, timestamp self.do_snapshot() self.do_timestamp() @@ -177,8 +205,6 @@ def submit_role(self, role: str, data: bytes) -> bool: if not targetpath.startswith(f"{role}/"): raise ValueError(f"targets allowed under {role}/ only") - self.targets().verify_delegate(role, md.signed_bytes, md.signatures) - if md.signed.version != self.targets(role).version + 1: raise ValueError("Invalid version {md.signed.version}") @@ -186,10 +212,19 @@ def submit_role(self, role: str, data: bytes) -> bool: logger.info("Failed to add new version for %s: %s", role, e) return False + # Check that we only write verified metadata + vr = self._get_verification_result(role, md) + if not vr: + logger.info("Role %s failed to verify", role) + return False + + keyids = [keyid[:7] for keyid in vr.signed] + verify_str = f"verified with keys [{', '.join(keyids)}]" + logger.debug("Role %s v%d: %s", role, md.signed.version, verify_str) + # Checks passed: Add new delegated role version self.role_cache[role].append(md) self._targets_infos[f"{role}.json"].version = md.signed.version - logger.debug("%s v%d", role, md.signed.version) # To keep it simple, target content is generated from targetpath for targetpath in md.signed.targets: From f60fb4abc8b20063bb48cad1e9552af949f054cd Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 13:51:28 +0200 Subject: [PATCH 08/13] Metadata API: Tweak get_root_verification_result args Change the "other" argument to optional "previous" and handle the None case in code. Signed-off-by: Jussi Kukkonen --- examples/repository/_simplerepo.py | 6 ++---- tuf/api/metadata.py | 18 ++++++++++++------ 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/examples/repository/_simplerepo.py b/examples/repository/_simplerepo.py index 668914622e..17ce7a0f47 100644 --- a/examples/repository/_simplerepo.py +++ b/examples/repository/_simplerepo.py @@ -99,11 +99,9 @@ def _get_verification_result( if role == Root.type: assert isinstance(md.signed, Root) root = self.root() - if root.version == 0: - # special case first root - root = md.signed + previous = root if root.version > 0 else None return md.signed.get_root_verification_result( - root, md.signed_bytes, md.signatures + previous, md.signed_bytes, md.signatures ) if role in [Timestamp.type, Snapshot.type, Targets.type]: delegator: Signed = self.root() diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 23bbce64cd..8ed7cbf31f 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -965,30 +965,36 @@ def get_key(self, keyid: str) -> Key: # noqa: D102 def get_root_verification_result( self, - other: "Root", + previous: Optional["Root"], payload: bytes, signatures: Dict[str, Signature], ) -> RootVerificationResult: """Return signature threshold verification result for two root roles. - Verify root metadata with two roles (the root role from `self` and - `other`). If you have only one role (in the case of root v1) you can - provide the same Root as both `self` and `other`. + Verify root metadata with two roles (`self` and optionally `previous`). + + If the repository has no root role versions yet, `previous` can be left + None. In all other cases, `previous` must be the previous version of + the Root. NOTE: Unlike `verify_delegate()` this method does not raise, if the root metadata is not fully verified. Args: - other: The other `Root` to verify payload with + previous: The previous `Root` to verify payload with, or None payload: Signed payload bytes for root signatures: Signatures over payload bytes Raises: ValueError: no delegation was found for ``delegated_role``. """ + + if previous is None: + previous = self + return RootVerificationResult( self.get_verification_result(Root.type, payload, signatures), - other.get_verification_result(Root.type, payload, signatures), + previous.get_verification_result(Root.type, payload, signatures), ) From 42d3a75787954f5dce5a2815edf7b7879ed93533 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 13:56:57 +0200 Subject: [PATCH 09/13] Metadata API: Improve docs for RootVerificationResult Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 8ed7cbf31f..99d5756bc1 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -653,8 +653,8 @@ class VerificationResult: Attributes: threshold: Number of required signatures. - signed: dict of keyid:Key containing the keys that have signed. - unsigned: dict of keyid:Key containing the keys that have not signed. + signed: dict of keyid to Key, containing keys that have signed. + unsigned: dict of keyid to Key, containing keys that have not signed. """ threshold: int @@ -675,8 +675,8 @@ class RootVerificationResult: """Signature verification result for root metadata. Root must be verified by itself and the previous root version. This - dataclass represents that combination. For the edge case of first version - of root, the same VerificationResult can be used twice. + dataclass represents both results. For the edge case of first version + of root, these underlying results are identical. Note that `signed` and `unsigned` correctness requires the underlying VerificationResult keys to not conflict (no reusing the same keyid for From b158c0852d91296f23877202c1f16a5e24ccbcbd Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 14:34:07 +0200 Subject: [PATCH 10/13] Metadata API: Make sanity checks in root verification Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 99d5756bc1..2d1129aa27 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -986,11 +986,16 @@ def get_root_verification_result( signatures: Signatures over payload bytes Raises: - ValueError: no delegation was found for ``delegated_role``. + ValueError: no delegation was found for ``root`` or given Root + versions are not sequential. """ if previous is None: previous = self + elif self.version != previous.version + 1: + versions = f"v{previous.version} and v{self.version}" + raise ValueError( + f"Expected sequential root versions, got {versions}.") return RootVerificationResult( self.get_verification_result(Root.type, payload, signatures), From 161c3e35adfe7a27f98279683c0c3e16d00e7ee3 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 15:01:46 +0200 Subject: [PATCH 11/13] Metadata API: Add VerificationResult.missing This is helper to tell how many signatures are still required. Also change the order of Roots given to RootVerificationResult (this way first is version N, second is version N+1). Signed-off-by: Jussi Kukkonen --- tuf/api/metadata.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 2d1129aa27..e64b2da9bc 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -669,6 +669,11 @@ def verified(self) -> bool: """True if threshold of signatures is met.""" return len(self.signed) >= self.threshold + @property + def missing(self) -> int: + """Number of additional signatures required to reach threshold.""" + return max(0, self.threshold - len(self.signed)) + @dataclass class RootVerificationResult: @@ -995,11 +1000,12 @@ def get_root_verification_result( elif self.version != previous.version + 1: versions = f"v{previous.version} and v{self.version}" raise ValueError( - f"Expected sequential root versions, got {versions}.") + f"Expected sequential root versions, got {versions}." + ) return RootVerificationResult( - self.get_verification_result(Root.type, payload, signatures), previous.get_verification_result(Root.type, payload, signatures), + self.get_verification_result(Root.type, payload, signatures), ) From bfea673893a6e4ed6b96e3ca2393111439d24e56 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 15:10:48 +0200 Subject: [PATCH 12/13] tests: Update the root verification tests Change tests so the previous root version is what the code expects. Signed-off-by: Jussi Kukkonen --- tests/test_api.py | 51 +++++++++++++++++++++++++++++------------------ 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/tests/test_api.py b/tests/test_api.py index ad3da15928..616425c259 100755 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -54,7 +54,7 @@ logger = logging.getLogger(__name__) -# pylint: disable=too-many-public-methods +# pylint: disable=too-many-public-methods,too-many-statements class TestMetadata(unittest.TestCase): """Tests for public API of all classes in 'tuf/api/metadata.py'.""" @@ -551,30 +551,43 @@ def test_root_get_root_verification_result(self) -> None: priv_key4 = self.keystore[Snapshot.type] - # other_root is only used as the other verifying role - other_root: Metadata[Root] = deepcopy(root) + # Test: Verify with no previous root version + result = root.signed.get_root_verification_result( + None, root.signed_bytes, root.signatures + ) + self.assertTrue(result) + self.assertEqual(result.signed, {key1_id: key1}) + self.assertEqual(result.unsigned, {}) + + # Test: Verify with other root that is not version N-1 + prev_root: Metadata[Root] = deepcopy(root) + with self.assertRaises(ValueError): + result = root.signed.get_root_verification_result( + prev_root.signed, root.signed_bytes, root.signatures + ) - # Test: Verify with two roles that are the same + # Test: Verify with previous root + prev_root.signed.version -= 1 result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertTrue(result) self.assertEqual(result.signed, {key1_id: key1}) self.assertEqual(result.unsigned, {}) - # Test: Add a signer to other root (threshold still 1) - other_root.signed.add_key(key2, Root.type) + # Test: Add a signer to previous root (threshold still 1) + prev_root.signed.add_key(key2, Root.type) result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertTrue(result) self.assertEqual(result.signed, {key1_id: key1}) self.assertEqual(result.unsigned, {key2_id: key2}) - # Test: Increase threshold in other root - other_root.signed.roles[Root.type].threshold += 1 + # Test: Increase threshold in previous root + prev_root.signed.roles[Root.type].threshold += 1 result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertFalse(result) self.assertEqual(result.signed, {key1_id: key1}) @@ -583,7 +596,7 @@ def test_root_get_root_verification_result(self) -> None: # Test: Sign root with both keys root.sign(SSlibSigner(priv_key2), append=True) result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertTrue(result) self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) @@ -592,25 +605,25 @@ def test_root_get_root_verification_result(self) -> None: # Test: Sign root with an unrelated key root.sign(SSlibSigner(priv_key4), append=True) result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertTrue(result) self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) self.assertEqual(result.unsigned, {}) - # Test: Remove key1 from other root - other_root.signed.revoke_key(key1_id, Root.type) + # Test: Remove key1 from previous root + prev_root.signed.revoke_key(key1_id, Root.type) result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertFalse(result) self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) self.assertEqual(result.unsigned, {}) - # Test: Lower threshold in other root - other_root.signed.roles[Root.type].threshold -= 1 + # Test: Lower threshold in previous root + prev_root.signed.roles[Root.type].threshold -= 1 result = root.signed.get_root_verification_result( - other_root.signed, root.signed_bytes, root.signatures + prev_root.signed, root.signed_bytes, root.signatures ) self.assertTrue(result) self.assertEqual(result.signed, {key1_id: key1, key2_id: key2}) From 14edf3d044b8c36d43d8271eac2880c2adc1cf2e Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Mon, 5 Feb 2024 15:26:31 +0200 Subject: [PATCH 13/13] tests: Add VerificationResult tests Signed-off-by: Jussi Kukkonen --- tests/test_api.py | 43 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/test_api.py b/tests/test_api.py index 616425c259..bf0b960623 100755 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -41,12 +41,14 @@ Metadata, MetaFile, Root, + RootVerificationResult, Signature, Snapshot, SuccinctRoles, TargetFile, Targets, Timestamp, + VerificationResult, ) from tuf.api.serialization import DeserializationError, SerializationError from tuf.api.serialization.json import JSONSerializer @@ -470,6 +472,47 @@ def test_signed_verify_delegate(self) -> None: Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures ) + def test_verification_result(self) -> None: + vr = VerificationResult(3, {"a": None}, {"b": None}) + self.assertEqual(vr.missing, 2) + self.assertFalse(vr.verified) + self.assertFalse(vr) + + # Add a signature + vr.signed["c"] = None + self.assertEqual(vr.missing, 1) + self.assertFalse(vr.verified) + self.assertFalse(vr) + + # Add last missing signature + vr.signed["d"] = None + self.assertEqual(vr.missing, 0) + self.assertTrue(vr.verified) + self.assertTrue(vr) + + # Add one more signature + vr.signed["e"] = None + self.assertEqual(vr.missing, 0) + self.assertTrue(vr.verified) + self.assertTrue(vr) + + def test_root_verification_result(self) -> None: + vr1 = VerificationResult(3, {"a": None}, {"b": None}) + vr2 = VerificationResult(1, {"c": None}, {"b": None}) + + vr = RootVerificationResult(vr1, vr2) + self.assertEqual(vr.signed, {"a": None, "c": None}) + self.assertEqual(vr.unsigned, {"b": None}) + self.assertFalse(vr.verified) + self.assertFalse(vr) + + vr1.signed["c"] = None + vr1.signed["f"] = None + self.assertEqual(vr.signed, {"a": None, "c": None, "f": None}) + self.assertEqual(vr.unsigned, {"b": None}) + self.assertTrue(vr.verified) + self.assertTrue(vr) + def test_signed_get_verification_result(self) -> None: # Setup: Load test metadata and keys root_path = os.path.join(self.repo_dir, "metadata", "root.json")