Fix noncompliant keyids (#338)

* signer: add hidden flag to fix keyid issue in -delegate

This modifies keyids in place: The key contents remain the same
* Unfortunately keys get shuffled around since they are sorted by keyid
* The result should be that same HW key will now sign the "new" keyids
* for root, the HW key will sign for both new keyids and old keyid
  (so that both root N and N+1 will reach threshold)

this command can be run on "root" and "targets" (and will fix all keyids
defined in that roles metadata). New versions of the delegated roles
will then be created to make sure they get signed with new keyids.

* signer: Allow multiple signatures

This makes it possible to sign the same metadata twice:
* currently this is only useful when fixing the keyid compliance issue
  in root (see #292). Basically the user will be asked to sign with
  both the keyid from root N+1 and the keyid from root N.
* there are clear use cases with one signer with multiple keys in
  future (think e.g. key rotation).

* signer: Also handle keyid fix during repo import

Since the import adds custom metadata into existing keys,
keyids become non-compliant. Run force_compliant_keyids() for
imported roles too.

This very annoyingly requires a special case in _sign(): basically
a heuristic that figures that we want to sign with "previous version"
root keys if the keyid of this legacy key can be calculated from
"current version" root key by just removing the custom metadata.

signer/tuf_on_ci_sign/_signer_repository.py

@@ -2,6 +2,7 @@
 
 """Internal repository module for TUF-on-CI signer tools"""
 
+import copy
 import filecmp
 import json
 import logging
@@ -739,27 +740,117 @@ def status(self, rolename: str) -> str:
 
         return "\n".join(output)
 
+    def _get_legacy_root_key(self, key: Key) -> Key | None:
+        # calculate keyid _without custom metadata_, then lookup key from known good
+        # root keys. This is useful in import situation where the legacy key does not
+        # have the custom metadata but is otherwise the same key
+        def _calculate_keyid(key: Key) -> str:
+            data: bytes = encode_canonical(key.to_dict()).encode()
+            hasher = digest("sha256")
+            hasher.update(data)
+            return hasher.hexdigest()
+
+        test_key = copy.deepcopy(key)
+        del test_key.unrecognized_fields["x-tuf-on-ci-keyowner"]
+        legacy_keyid = _calculate_keyid(test_key)
+        return self._known_good_root().keys.get(legacy_keyid)
+
     def sign(self, rolename: str):
         """Sign without payload changes"""
         md = self.open(rolename)
+        signing_keys: dict[str, Key] = {}
         for key in self._get_keys(rolename):
             keyowner = key.unrecognized_fields["x-tuf-on-ci-keyowner"]
             if keyowner == self.user.name:
-                self._sign(rolename, md, key)
-                self._write(rolename, md)
-                return
+                signing_keys[key.keyid] = key
 
-        # Root is eligible to sign current root if the signer was valid
-        # in previous version
         if rolename == "root":
+            # special case for import: if the same key was used with different keyid
+            # we want to sign with that keyid too
+            for key in list(signing_keys.values()):
+                legacy_key = self._get_legacy_root_key(key)
+                if legacy_key:
+                    signing_keys[legacy_key.keyid] = legacy_key
+
+            # user is also eligible to sign current root if they were a signer
+            # in previous version
             for key in self._get_keys(rolename, True):
                 keyowner = key.unrecognized_fields["x-tuf-on-ci-keyowner"]
                 if keyowner == self.user.name:
-                    self._sign(rolename, md, key)
-                    self._write(rolename, md)
-                    return
+                    signing_keys[key.keyid] = key
+
+        if not signing_keys:
+            raise ValueError(f"{rolename} signing key for {self.user.name} not found")
+
+        for key in signing_keys.values():
+            self._sign(rolename, md, key)
+        self._write(rolename, md)
+
+    def force_compliant_keyids(self, rolename: str) -> bool:
+        """Make all keyids defined in rolename spec compliant
+
+        This is a hidden feature to fix issue #294. It updates all keyids defined
+        in the role so that they are spec compliant: this means changes in the
+        delegated roles that use the keyids and requires resigning the metadata
+        of those roles.
+
+        Requires resigning with care: Root signatures should be duplicated for new and
+        old keyids."""
+
+        def _calculate_keyid(key: Key) -> str:
+            data: bytes = encode_canonical(key.to_dict()).encode()
+            hasher = digest("sha256")
+            hasher.update(data)
+            return hasher.hexdigest()
+
+        changed = False
+        delegates = set()
+        if rolename == "root":
+            with self.edit_root() as root:
+                for key in list(root.keys.values()):
+                    compliant_keyid = _calculate_keyid(key)
+                    if key.keyid == compliant_keyid:
+                        continue
+                    # Update keyid in all roles
+                    for rolename, role in root.roles.items():
+                        for i, id in enumerate(role.keyids):
+                            if id == key.keyid:
+                                role.keyids[i] = compliant_keyid
+                                if rolename == "targets":
+                                    delegates.add(rolename)
+
+                    # update the actual key
+                    del root.keys[key.keyid]
+                    key.keyid = compliant_keyid
+                    root.keys[key.keyid] = key
+
+                    changed = True
+        elif rolename == "targets":
+            with self.edit_targets() as targets:
+                if not targets.delegations or not targets.delegations.roles:
+                    raise AbortEdit
+                for key in list(targets.delegations.keys.values()):
+                    compliant_keyid = _calculate_keyid(key)
+                    if key.keyid == compliant_keyid:
+                        continue
+                    # Update keyid in the key and all roles
+                    for rolename, role in targets.delegations.roles.items():
+                        for i, id in enumerate(role.keyids):
+                            if id == key.keyid:
+                                role.keyids[i] = compliant_keyid
+                                delegates.add(rolename)
+                    # update the actual key
+                    del targets.delegations.keys[key.keyid]
+                    key.keyid = compliant_keyid
+                    targets.delegations.keys[key.keyid] = key
+                    changed = True
+
+        for delegate in delegates:
+            # Force resigning of delegates
+            with self.edit_targets(delegate):
+                pass
 
-        raise ValueError(f"{rolename} signing key for {self.user.name} not found")
+        return changed
 
 
 def build_paths(rolename: str, depth: int) -> list[str]:

signer/tuf_on_ci_sign/delegate.py

@@ -338,9 +338,16 @@ def _update_offline_role(repo: SignerRepository, role: str) -> bool:
 @click.version_option()
 @click.option("-v", "--verbose", count=True, default=0)
 @click.option("--push/--no-push", default=True)
+@click.option("--force-compliant-keyids", hidden=True, is_flag=True)
 @click.argument("event-name", metavar="SIGNING-EVENT")
 @click.argument("role", required=False)
-def delegate(verbose: int, push: bool, event_name: str, role: str | None):
+def delegate(
+    verbose: int,
+    push: bool,
+    force_compliant_keyids: bool,
+    event_name: str,
+    role: str | None,
+):
     """Tool for modifying TUF-on-CI delegations."""
     logging.basicConfig(level=logging.WARNING - verbose * 10)
 
@@ -357,7 +364,9 @@ def delegate(verbose: int, push: bool, event_name: str, role: str | None):
             if role is None:
                 role = click.prompt(bold("Enter name of role to modify"))
 
-            if role in ["timestamp", "snapshot"]:
+            if force_compliant_keyids:
+                changed = repo.force_compliant_keyids(role)
+            elif role in ["timestamp", "snapshot"]:
                 changed = _update_online_roles(repo)
             else:
                 changed = _update_offline_role(repo, role)
@@ -381,7 +390,6 @@ def delegate(verbose: int, push: bool, event_name: str, role: str | None):
                 git_expect(
                     ["commit", "-m", f"Signed by {user_config.name}", "--signoff"]
                 )
-
             if push:
                 push_changes(user_config, event_name, msg)
             else:

signer/tuf_on_ci_sign/import_repo.py

@@ -162,6 +162,7 @@ def import_repo(verbose: int, push: bool, event_name: str, import_file: str | No
                     ok = _update_keys(root.keys, role_data) and ok
                     if not ok:
                         raise AbortEdit("Missing values")
+
             else:
                 with repo.edit_targets(rolename) as targets:
                     ok = _update_expiry(targets, role_data) and ok
@@ -172,6 +173,10 @@ def import_repo(verbose: int, push: bool, event_name: str, import_file: str | No
                     if not ok:
                         raise AbortEdit("Missing values")
 
+        # we have updated keys defined in root/targets: make sure keyids are compliant
+        repo.force_compliant_keyids("root")
+        repo.force_compliant_keyids("targets")
+
         if not ok:
             print("Error: Undefined values found. please save this in a file,")
             print("fill in the values and use the file as import-file argument:\n")