Merge pull request #503 from ronantakizawa/ronantakizawa/1+yubikeys

feat: Use multiple yubikeys at same time

.github/workflows/ci.yml

@@ -199,4 +199,4 @@ jobs:
         run: |
           gh release upload ${{github.event.release.tag_name}} dist/taf-macos
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

CHANGELOG.md

@@ -10,9 +10,10 @@ and this project adheres to [Semantic Versioning][semver].
 ### Added
 
 - Added yubikey_present parameter to keys description (Can be specified when generating keys) ([508])
+- Added new tests to test out of sync repositories and manual updates ([488], [504])
+- Use multiple yubikeys at same time (Key generation, repo generation, target signing) ([503])
 - Removed 2048-bit key restriction [494]
 - Allow for the displaying of varied levels of log and debug information based on the verbosity level ([493])
-- Added new tests to test out of sync repositories and manual updates ([488], [504])
 - Update when auth repo's top commit is behind last validated commit [490]
 - Added lazy loading to CLI [481]
 - Testing repositories with dependencies ([479], [487])
@@ -42,6 +43,7 @@ and this project adheres to [Semantic Versioning][semver].
 
 [508]: https://github.com/openlawlibrary/taf/pull/508
 [504]: https://github.com/openlawlibrary/taf/pull/504
+[503]: https://github.com/openlawlibrary/taf/pull/503
 [494]: https://github.com/openlawlibrary/taf/pull/494
 [493]: https://github.com/openlawlibrary/taf/pull/493
 [490]: https://github.com/openlawlibrary/taf/pull/490

setup.py

@@ -33,7 +33,7 @@
     "jinja2==3.1.*",
 ]
 
-yubikey_require = ["yubikey-manager==5.1.*"]
+yubikey_require = ["yubikey-manager==5.5.*"]
 
 # Determine the appropriate version of pygit2 based on the Python version
 if sys.version_info >= (3, 11):

taf/api/yubikey.py

@@ -141,6 +141,7 @@ def setup_signing_yubikey(
         "WARNING - this will delete everything from the inserted key. Proceed?"
     ):
         return
+
     _, serial_num = yk.yubikey_prompt(
         "new Yubikey",
         creating_new_key=True,

taf/keys.py

@@ -576,8 +576,7 @@ def _setup_yubikey(
 def _load_and_verify_yubikey(
     yubikeys: Optional[Dict], role_name: str, key_name: str, public_key
 ) -> bool:
-    if not click.confirm(f"Sign using {key_name} Yubikey?"):
-        return False
+    # Attempt to load and verify the YubiKey
     while True:
         yk_public_key, _ = yk.yubikey_prompt(
             key_name,
@@ -588,10 +587,11 @@ def _load_and_verify_yubikey(
             loaded_yubikeys=yubikeys,
             pin_confirm=True,
             pin_repeat=True,
+            expected_public_key=public_key,
         )
 
-        if yk_public_key["keyid"] != public_key["keyid"]:
-            print("Public key of the inserted key is not equal to the specified one.")
-            if not click.confirm("Try again?"):
-                return False
-        return True
+        # Check if the public key matches
+        if yk_public_key["keyid"] == public_key["keyid"]:
+            return True
+
+        print("Public key of the inserted key does not match the specified one.")

taf/repository_tool.py

@@ -7,6 +7,7 @@
 from functools import partial, reduce
 from pathlib import Path
 from typing import Dict
+from ykman.device import list_all_devices
 
 import securesystemslib
 import tuf.roledb
@@ -139,34 +140,59 @@ def root_signature_provider(signature_dict, key_id, _key, _data):
 
 def yubikey_signature_provider(name, key_id, key, data):  # pylint: disable=W0613
     """
-    A signatures provider which asks the user to insert a yubikey
-    Useful if several yubikeys need to be used at the same time
+    A signature provider which asks the user to insert a YubiKey.
+    Useful if several YubiKeys need to be used at the same time.
     """
     from binascii import hexlify
 
-    def _check_key_and_get_pin(expected_key_id):
+    def _check_key_and_get_pin_legacy(expected_key_id):
         try:
             inserted_key = yk.get_piv_public_key_tuf()
             if expected_key_id != inserted_key["keyid"]:
-                return None
+                return None, None
             serial_num = yk.get_serial_num(inserted_key)
             pin = yk.get_key_pin(serial_num)
             if pin is None:
                 pin = yk.get_and_validate_pin(name)
-            return pin
+            return pin, serial_num
         except Exception:
-            return None
+            return None, None
+
+    def _check_key_and_get_pin(expected_key_id):
+        devices = list_all_devices()
+        if len(devices) < 2:
+            # Use the legacy method if there is only one YubiKey connected
+            return _check_key_and_get_pin_legacy(expected_key_id)
+
+        for _, info in devices:
+            serial_num = info.serial
+            try:
+                # Check if the YubiKey is loaded with the correct key
+                inserted_key = yk.get_piv_public_key_tuf(serial=serial_num)
+                if inserted_key["keyid"] == expected_key_id:
+                    pin = yk.get_key_pin(serial_num)
+                    if pin is None:
+                        pin = yk.get_and_validate_pin(name, serial=serial_num)
+                    return pin, serial_num
+                else:
+                    # If the key doesn't match, continue to the next device
+                    continue
+
+            except Exception as e:
+                print(f"Error with YubiKey {serial_num}: {str(e)}")
+                return None, None
+
+        # If no suitable YubiKey is found after checking all devices
+        return None, None
 
     while True:
-        # check if the needed YubiKey is inserted before asking the user to do so
-        # this allows us to use this signature provider inside an automated process
-        # assuming that all YubiKeys needed for signing are inserted
-        pin = _check_key_and_get_pin(key_id)
+        # Check if the needed YubiKey is inserted before asking the user to do so.
+        pin, serial = _check_key_and_get_pin(key_id)
         if pin is not None:
             break
-        input(f"\nInsert {name} and press enter")
+        input(f"\nInsert {name} YubiKey and press enter")
 
-    signature = yk.sign_piv_rsa_pkcs1v15(data, pin)
+    signature = yk.sign_piv_rsa_pkcs1v15(data, pin, serial=serial)
     return {"keyid": key_id, "sig": hexlify(signature).decode()}
 
 

taf/tests/test_repository_tool/test_repository_tool.py

@@ -7,6 +7,7 @@
 
 import taf.exceptions
 import taf.yubikey as yk
+
 from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME
 from taf.tests import TEST_WITH_REAL_YK
 from taf.tests.yubikey_utils import VALID_PIN

taf/tools/yubikey/__init__.py

@@ -3,6 +3,7 @@
 from taf.api.yubikey import export_yk_certificate, export_yk_public_pem, get_yk_roles, setup_signing_yubikey, setup_test_yubikey
 from taf.exceptions import YubikeyError
 from taf.repository_utils import find_valid_repository
+from taf.yubikey import list_connected_yubikeys
 from taf.tools.cli import catch_cli_exception
 
 
@@ -71,14 +72,22 @@ def setup_test_key_command():
         WARNING - this will reset the inserted key.""")
     @click.argument("key-path")
     def setup_test_key(key_path):
-        setup_test_yubikey(key_path,key_size=2048)
+        setup_test_yubikey(key_path)
     return setup_test_key
 
 
+def list_key_command():
+    @click.command(help="List All Connected Keys and their information")
+    def list_keys():
+        list_connected_yubikeys()
+    return list_keys
+
+
 def attach_to_group(group):
     group.add_command(check_pin_command(), name='check-pin')
     group.add_command(export_pub_key_command(), name='export-pub-key')
     group.add_command(get_roles_command(), name='get-roles')
     group.add_command(export_certificate_command(), name='export-certificate')
     group.add_command(setup_signing_key_command(), name='setup-signing-key')
     group.add_command(setup_test_key_command(), name='setup-test-key')
+    group.add_command(list_key_command(), name='list-key')