diff --git a/tests/test_trusted_metadata_set.py b/tests/test_trusted_metadata_set.py index fd59635ed8..f2b93d0890 100644 --- a/tests/test_trusted_metadata_set.py +++ b/tests/test_trusted_metadata_set.py @@ -149,6 +149,19 @@ def test_update(self) -> None: self.assertTrue(count, 6) + def test_contains(self) -> None: + self.assertFalse(self.trusted_set.contains("role1", Targets.type)) + + self.trusted_set.update_timestamp(self.metadata[Timestamp.type]) + self.trusted_set.update_snapshot(self.metadata[Snapshot.type]) + self.trusted_set.update_targets(self.metadata[Targets.type]) + self.trusted_set.update_delegated_targets( + self.metadata["role1"], "role1", Targets.type + ) + + self.assertTrue(self.trusted_set.contains("role1", Targets.type)) + self.assertFalse(self.trusted_set.contains("role1", "other-parent")) + def test_update_metadata_output(self) -> None: timestamp = self.trusted_set.update_timestamp( self.metadata["timestamp"] diff --git a/tests/test_updater_delegation_graphs.py b/tests/test_updater_delegation_graphs.py index 536bb13a2d..832f0dbc32 100644 --- a/tests/test_updater_delegation_graphs.py +++ b/tests/test_updater_delegation_graphs.py @@ -330,6 +330,43 @@ def test_invalid_metadata(self, test_data: DelegationsTestCase) -> None: finally: self.teardown_subtest() + def test_shared_delegation_signature_check(self) -> None: + """Test that a role delegated by multiple parents is signature-checked + for each delegator link. + """ + test_case = DelegationsTestCase( + delegations=[ + TestDelegation("targets", "parent-a", paths=["path-a"]), + TestDelegation("targets", "parent-b", paths=["path-b"]), + TestDelegation("parent-a", "release", paths=["path-a"]), + TestDelegation("parent-b", "release", paths=["path-b"]), + ], + target_files=[ + TestTarget("release", b"content-a", "path-a"), + TestTarget("release", b"malicious-content-b", "path-b"), + ], + ) + self._init_repo(test_case) + + # release metadata is only signed by parent-a + first_keyid = next(iter(self.sim.signers["release"].keys())) + self.sim.signers["release"] = { + first_keyid: self.sim.signers["release"][first_keyid] + } + self.sim.update_snapshot() + + # Client step 1: Get path-a targetinfo via parent-a -> release + # This primes TrustedMetadataSet to contain release metadata + updater = self._init_updater() + target_a = updater.get_targetinfo("path-a") + self.assertIsNotNone(target_a) + + # Client step 2: Get path-b targetinfo via parent-b -> release + # Must fail because release while loaded in TrustedMetadataSet, is not signed + # by parent-b's keys + with self.assertRaises(UnsignedMetadataError): + updater.get_targetinfo("path-b") + def test_safely_encoded_rolenames(self) -> None: """Test that delegated roles names are safely encoded in the filenames and URLs. diff --git a/tuf/ngclient/_internal/trusted_metadata_set.py b/tuf/ngclient/_internal/trusted_metadata_set.py index 689eef01de..b48c5a5d3c 100644 --- a/tuf/ngclient/_internal/trusted_metadata_set.py +++ b/tuf/ngclient/_internal/trusted_metadata_set.py @@ -115,6 +115,7 @@ def __init__(self, root_data: bytes, envelope_type: EnvelopeType): error type and content will contain more details. """ self._trusted_set: dict[str, Signed] = {} + self._trusted_delegations: set[tuple[str, str]] = set() self.reference_time = datetime.datetime.now(datetime.timezone.utc) if envelope_type is EnvelopeType.SIMPLE: @@ -127,6 +128,17 @@ def __init__(self, root_data: bytes, envelope_type: EnvelopeType): logger.debug("Updating initial trusted root") self._load_trusted_root(root_data) + def contains(self, role: str, delegator: str) -> bool: + """Check if ``role`` is in ``TrustedMetadataSet`` and was verified + against ``delegator``. + """ + return (delegator, role) in self._trusted_delegations + + def _add(self, role: str, signed: Signed, delegator: str) -> None: + """Add role to trusted set, keep track of delegator(s)""" + self._trusted_set[role] = signed + self._trusted_delegations.add((delegator, role)) + def __getitem__(self, role: str) -> Signed: """Return current ``Signed`` for ``role``.""" return self._trusted_set[role] @@ -196,7 +208,7 @@ def update_root(self, data: bytes) -> Root: # Verify that new root is signed by itself new_root.verify_delegate(Root.type, new_root_bytes, new_root_signatures) - self._trusted_set[Root.type] = new_root + self._add(Root.type, new_root, Root.type) logger.debug("Updated root v%d", new_root.version) return new_root @@ -259,7 +271,7 @@ def update_timestamp(self, data: bytes) -> Timestamp: # expiry not checked to allow old timestamp to be used for rollback # protection of new timestamp: expiry is checked in update_snapshot() - self._trusted_set[Timestamp.type] = new_timestamp + self._add(Timestamp.type, new_timestamp, Root.type) logger.debug("Updated timestamp v%d", new_timestamp.version) # timestamp is loaded: raise if it is not valid _final_ timestamp @@ -346,7 +358,7 @@ def update_snapshot( # expiry not checked to allow old snapshot to be used for rollback # protection of new snapshot: it is checked when targets is updated - self._trusted_set[Snapshot.type] = new_snapshot + self._add(Snapshot.type, new_snapshot, Root.type) logger.debug("Updated snapshot v%d", new_snapshot.version) # snapshot is loaded, but we raise if it's not valid _final_ snapshot @@ -434,7 +446,7 @@ def update_delegated_targets( if new_delegate.is_expired(self.reference_time): raise exceptions.ExpiredMetadataError(f"New {role_name} is expired") - self._trusted_set[role_name] = new_delegate + self._add(role_name, new_delegate, delegator_name) logger.debug("Updated %s v%d", role_name, version) return new_delegate @@ -450,7 +462,7 @@ def _load_trusted_root(self, data: bytes) -> None: ) new_root.verify_delegate(Root.type, new_root_bytes, new_root_signatures) - self._trusted_set[Root.type] = new_root + self._add(Root.type, new_root, Root.type) logger.debug("Loaded trusted root v%d", new_root.version) diff --git a/tuf/ngclient/updater.py b/tuf/ngclient/updater.py index a253b18d4c..17e7ab8a53 100644 --- a/tuf/ngclient/updater.py +++ b/tuf/ngclient/updater.py @@ -462,8 +462,8 @@ def _load_snapshot(self) -> None: def _load_targets(self, role: str, parent_role: str) -> Targets: """Load local (and if needed remote) metadata for ``role``.""" - # Avoid loading 'role' more than once during "get_targetinfo" - if role in self._trusted_set: + # Avoid loading 'role' more than once for the same parent_role + if self._trusted_set.contains(role, parent_role): return cast("Targets", self._trusted_set[role]) try: