diff --git a/examples/repo_example/hashed_bin_delegation.py b/examples/repo_example/hashed_bin_delegation.py index c8bc3b34..f287afe1 100644 --- a/examples/repo_example/hashed_bin_delegation.py +++ b/examples/repo_example/hashed_bin_delegation.py @@ -162,6 +162,7 @@ def find_hash_bin(path: str) -> str: # 10-17 10 11 12 13 14 15 16 17 # ... ... # f8-ff f8 f9 fa fb fc fd fe ff +assert roles["bins"].signed.delegations.roles is not None for bin_n_name, bin_n_hash_prefixes in generate_hash_bins(): # Update delegating targets role (bins) with delegation details for each # delegated targets role (bin_n). diff --git a/tests/repository_simulator.py b/tests/repository_simulator.py index ae1ad3e6..607bb47e 100644 --- a/tests/repository_simulator.py +++ b/tests/repository_simulator.py @@ -354,9 +354,17 @@ def add_delegation( else: delegator = self.md_delegates[delegator_name].signed + if ( + delegator.delegations is not None + and delegator.delegations.succinct_roles is not None + ): + raise ValueError("Can't add a role when succinct_roles is used") + # Create delegation if delegator.delegations is None: - delegator.delegations = Delegations({}, {}) + delegator.delegations = Delegations({}, roles={}) + + assert delegator.delegations.roles is not None # put delegation last by default delegator.delegations.roles[role.name] = role diff --git a/tests/test_api.py b/tests/test_api.py index afe947a7..2770bbaf 100755 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -495,6 +495,7 @@ def test_targets_key_api(self) -> None: } ) assert isinstance(targets.delegations, Delegations) + assert isinstance(targets.delegations.roles, Dict) targets.delegations.roles["role2"] = delegated_role key_dict = { diff --git a/tests/test_metadata_eq_.py b/tests/test_metadata_eq_.py index 88a87c75..a3b3f9fd 100644 --- a/tests/test_metadata_eq_.py +++ b/tests/test_metadata_eq_.py @@ -169,6 +169,7 @@ def test_delegations_eq_roles_reversed_order(self) -> None: # Create a second delegations obj with reversed roles order delegations_2 = copy.deepcopy(delegations) # In python3.7 we need to cast to a list and then reverse. + assert isinstance(delegations.roles, dict) delegations_2.roles = dict(reversed(list(delegations.roles.items()))) # Both objects are not the equal because of delegated roles order. diff --git a/tests/test_metadata_serialization.py b/tests/test_metadata_serialization.py index 800b100a..cec4fd08 100644 --- a/tests/test_metadata_serialization.py +++ b/tests/test_metadata_serialization.py @@ -491,6 +491,13 @@ def test_invalid_succinct_roles_serialization(self, test_data: str) -> None: {"keyids": ["keyid1"], "name": "b", "terminating": true, "paths": ["fn1"], "threshold": 3}, \ {"keyids": ["keyid2"], "name": "root", "terminating": true, "paths": ["fn2"], "threshold": 4} ] \ }', + "roles and succinct_roles set": '{"keys": { \ + "keyid1" : {"keytype": "rsa", "scheme": "rsassa-pss-sha256", "keyval": {"public": "foo"}}, \ + "keyid2" : {"keytype": "ed25519", "scheme": "ed25519", "keyval": {"public": "bar"}}}, \ + "roles": [ \ + {"keyids": ["keyid"], "name": "a", "terminating": true, "paths": ["fn1"], "threshold": 3}, \ + {"keyids": ["keyid2"], "name": "b", "terminating": true, "paths": ["fn2"], "threshold": 4} ], \ + "succinct_roles": {"keyids": ["keyid"], "threshold": 1, "bit_length": 8, "name_prefix": "foo"}}', } @utils.run_sub_tests_with_dataset(invalid_delegations) @@ -502,13 +509,17 @@ def test_invalid_delegation_serialization( Delegations.from_dict(case_dict) valid_delegations: utils.DataSet = { - "all": '{"keys": { \ + "with roles": '{"keys": { \ "keyid1" : {"keytype": "rsa", "scheme": "rsassa-pss-sha256", "keyval": {"public": "foo"}}, \ "keyid2" : {"keytype": "ed25519", "scheme": "ed25519", "keyval": {"public": "bar"}}}, \ "roles": [ \ {"keyids": ["keyid"], "name": "a", "terminating": true, "paths": ["fn1"], "threshold": 3}, \ {"keyids": ["keyid2"], "name": "b", "terminating": true, "paths": ["fn2"], "threshold": 4} ] \ }', + "with succinct_roles": '{"keys": { \ + "keyid1" : {"keytype": "rsa", "scheme": "rsassa-pss-sha256", "keyval": {"public": "foo"}}, \ + "keyid2" : {"keytype": "ed25519", "scheme": "ed25519", "keyval": {"public": "bar"}}}, \ + "succinct_roles": {"keyids": ["keyid"], "threshold": 1, "bit_length": 8, "name_prefix": "foo"}}', "unrecognized field": '{"keys": {"keyid" : {"keytype": "rsa", "scheme": "rsassa-pss-sha256", "keyval": {"public": "foo"}}}, \ "roles": [ {"keyids": ["keyid"], "name": "a", "paths": ["fn1", "fn2"], "terminating": true, "threshold": 3} ], \ "foo": "bar"}', diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index fe858b9b..4029468e 100644 --- a/tuf/api/metadata.py +++ b/tuf/api/metadata.py @@ -413,7 +413,7 @@ def verify_delegate( """ # Find the keys and role in delegator metadata - role = None + role: Optional[Role] = None if isinstance(self.signed, Root): keys = self.signed.keys role = self.signed.roles.get(delegated_role) @@ -422,7 +422,13 @@ def verify_delegate( raise ValueError(f"No delegation found for {delegated_role}") keys = self.signed.delegations.keys - role = self.signed.delegations.roles.get(delegated_role) + if self.signed.delegations.roles is not None: + role = self.signed.delegations.roles.get(delegated_role) + elif self.signed.delegations.succinct_roles is not None: + if self.signed.delegations.succinct_roles.is_delegated_role( + delegated_role + ): + role = self.signed.delegations.succinct_roles else: raise TypeError("Call is valid only on delegator metadata") @@ -1592,9 +1598,17 @@ class Delegations: defines which keys are required to sign the metadata for a specific role. The roles order also defines the order that role delegations are considered during target searches. + succinct_roles: Contains succinct information about hash bin + delegations. Note that succinct roles is not a TUF specification + feature yet and setting `succinct_roles` to a value makes the + resulting metadata non-compliant. The metadata will not be accepted + as valid by specification compliant clients such as those built with + python-tuf <= 1.1.0. For more information see: https://github.com/theupdateframework/taps/blob/master/tap15.md unrecognized_fields: Dictionary of all attributes that are not managed by TUF Metadata API + Exactly one of ``roles`` and ``succinct_roles`` must be set. + Raises: ValueError: Invalid arguments. """ @@ -1602,18 +1616,24 @@ class Delegations: def __init__( self, keys: Dict[str, Key], - roles: Dict[str, DelegatedRole], + roles: Optional[Dict[str, DelegatedRole]] = None, + succinct_roles: Optional[SuccinctRoles] = None, unrecognized_fields: Optional[Dict[str, Any]] = None, ): self.keys = keys + if sum(1 for v in [roles, succinct_roles] if v is not None) != 1: + raise ValueError("One of roles and succinct_roles must be set") - for role in roles: - if not role or role in TOP_LEVEL_ROLE_NAMES: - raise ValueError( - "Delegated roles cannot be empty or use top-level role names" - ) + if roles is not None: + for role in roles: + if not role or role in TOP_LEVEL_ROLE_NAMES: + raise ValueError( + "Delegated roles cannot be empty or use top-level " + "role names" + ) self.roles = roles + self.succinct_roles = succinct_roles if unrecognized_fields is None: unrecognized_fields = {} @@ -1623,14 +1643,22 @@ def __eq__(self, other: Any) -> bool: if not isinstance(other, Delegations): return False - return ( + all_attributes_check = ( self.keys == other.keys - # Order of the delegated roles matters (see issue #1788). - and list(self.roles.items()) == list(other.roles.items()) and self.roles == other.roles + and self.succinct_roles == other.succinct_roles and self.unrecognized_fields == other.unrecognized_fields ) + if self.roles is not None and other.roles is not None: + all_attributes_check = ( + all_attributes_check + # Order of the delegated roles matters (see issue #1788). + and list(self.roles.items()) == list(other.roles.items()) + ) + + return all_attributes_check + @classmethod def from_dict(cls, delegations_dict: Dict[str, Any]) -> "Delegations": """Creates ``Delegations`` object from its json/dict representation. @@ -1642,25 +1670,59 @@ def from_dict(cls, delegations_dict: Dict[str, Any]) -> "Delegations": keys_res = {} for keyid, key_dict in keys.items(): keys_res[keyid] = Key.from_dict(keyid, key_dict) - roles = delegations_dict.pop("roles") - roles_res: Dict[str, DelegatedRole] = {} - for role_dict in roles: - new_role = DelegatedRole.from_dict(role_dict) - if new_role.name in roles_res: - raise ValueError(f"Duplicate role {new_role.name}") - roles_res[new_role.name] = new_role + roles = delegations_dict.pop("roles", None) + roles_res: Optional[Dict[str, DelegatedRole]] = None + + if roles is not None: + roles_res = {} + for role_dict in roles: + new_role = DelegatedRole.from_dict(role_dict) + if new_role.name in roles_res: + raise ValueError(f"Duplicate role {new_role.name}") + roles_res[new_role.name] = new_role + + succinct_roles_dict = delegations_dict.pop("succinct_roles", None) + succinct_roles_info = None + if succinct_roles_dict is not None: + succinct_roles_info = SuccinctRoles.from_dict(succinct_roles_dict) + # All fields left in the delegations_dict are unrecognized. - return cls(keys_res, roles_res, delegations_dict) + return cls(keys_res, roles_res, succinct_roles_info, delegations_dict) def to_dict(self) -> Dict[str, Any]: """Returns the dict representation of self.""" keys = {keyid: key.to_dict() for keyid, key in self.keys.items()} - roles = [role_obj.to_dict() for role_obj in self.roles.values()] - return { + res_dict: Dict[str, Any] = { "keys": keys, - "roles": roles, **self.unrecognized_fields, } + if self.roles is not None: + roles = [role_obj.to_dict() for role_obj in self.roles.values()] + res_dict["roles"] = roles + elif self.succinct_roles is not None: + res_dict["succinct_roles"] = self.succinct_roles.to_dict() + + return res_dict + + def get_roles_for_target( + self, target_filepath: str + ) -> Iterator[Tuple[str, bool]]: + """Given ``target_filepath`` get names and terminating status of all + delegated roles who are responsible for it. + + Args: + target_filepath: URL path to a target file, relative to a base + targets URL. + """ + if self.roles is not None: + for role in self.roles.values(): + if role.is_delegated_path(target_filepath): + yield role.name, role.terminating + + elif self.succinct_roles is not None: + # We consider all succinct_roles as terminating. + # For more information read TAP 15. + yield self.succinct_roles.get_role_for_target(target_filepath), True class TargetFile(BaseFile): @@ -1921,11 +1983,15 @@ def add_key(self, role: str, key: Key) -> None: ValueError: If there are no delegated roles or if ``role`` is not delegated by this Target. """ - if self.delegations is None or role not in self.delegations.roles: + if self.delegations is None: raise ValueError(f"Delegated role {role} doesn't exist") - if key.keyid not in self.delegations.roles[role].keyids: - self.delegations.roles[role].keyids.append(key.keyid) - self.delegations.keys[key.keyid] = key + + if self.delegations.roles is not None: + if role not in self.delegations.roles: + raise ValueError(f"Delegated role {role} doesn't exist") + if key.keyid not in self.delegations.roles[role].keyids: + self.delegations.roles[role].keyids.append(key.keyid) + self.delegations.keys[key.keyid] = key def remove_key(self, role: str, keyid: str) -> None: """Removes key from delegated role ``role`` and updates the delegations @@ -1939,13 +2005,18 @@ def remove_key(self, role: str, keyid: str) -> None: ValueError: If there are no delegated roles or if ``role`` is not delegated by this ``Target`` or if key is not used by ``role``. """ - if self.delegations is None or role not in self.delegations.roles: + if self.delegations is None: raise ValueError(f"Delegated role {role} doesn't exist") - if keyid not in self.delegations.roles[role].keyids: - raise ValueError(f"Key with id {keyid} is not used by {role}") - self.delegations.roles[role].keyids.remove(keyid) - for keyinfo in self.delegations.roles.values(): - if keyid in keyinfo.keyids: - return - del self.delegations.keys[keyid] + if self.delegations.roles is not None: + if role not in self.delegations.roles: + raise ValueError(f"Delegated role {role} doesn't exist") + if keyid not in self.delegations.roles[role].keyids: + raise ValueError(f"Key with id {keyid} is not used by {role}") + + self.delegations.roles[role].keyids.remove(keyid) + for keyinfo in self.delegations.roles.values(): + if keyid in keyinfo.keyids: + return + + del self.delegations.keys[keyid] diff --git a/tuf/ngclient/updater.py b/tuf/ngclient/updater.py index 914163e7..150d195a 100644 --- a/tuf/ngclient/updater.py +++ b/tuf/ngclient/updater.py @@ -438,17 +438,17 @@ def _preorder_depth_first_walk( child_roles_to_visit = [] # NOTE: This may be a slow operation if there are many # delegated roles. - for child_role in targets.delegations.roles.values(): - if child_role.is_delegated_path(target_filepath): - logger.debug("Adding child role %s", child_role.name) + for ( + child_name, + terminating, + ) in targets.delegations.get_roles_for_target(target_filepath): - child_roles_to_visit.append( - (child_role.name, role_name) - ) - if child_role.terminating: - logger.debug("Not backtracking to other roles") - delegations_to_visit = [] - break + logger.debug("Adding child role %s", child_name) + child_roles_to_visit.append((child_name, role_name)) + if terminating: + logger.debug("Not backtracking to other roles") + delegations_to_visit = [] + break # Push 'child_roles_to_visit' in reverse order of appearance # onto 'delegations_to_visit'. Roles are popped from the end of # the list.