Fix incorrect delegation in diamond-shape case

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"]

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.

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)
 
 

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: