From 43b9f15ddca976541431e601c02fb1096c26e976 Mon Sep 17 00:00:00 2001 From: Jussi Kukkonen Date: Wed, 5 Aug 2026 18:53:34 +0300 Subject: [PATCH] Fix incorrect delegation in diamond-shape case In a diamond delegation (where a role is delegated to from multiple delegating roles), if client has already verified a delegation through one branch and is now verifying the delegation through the other branch, it should still make sure the delegation is correctly signed via the second set of signers. The fix could be smaller and isloated in updater.py but I added the refactoring with `TrustedMetadataSet._add()` and `TrustedMetadataSet.contains()` to make this clearer. Fixes #2977 Signed-off-by: Jussi Kukkonen --- tests/test_trusted_metadata_set.py | 13 +++++++ tests/test_updater_delegation_graphs.py | 37 +++++++++++++++++++ .../_internal/trusted_metadata_set.py | 22 ++++++++--- tuf/ngclient/updater.py | 4 +- 4 files changed, 69 insertions(+), 7 deletions(-) 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: