From 1e1058d5005bc622f2ad20f75f22cdc24f4a84b9 Mon Sep 17 00:00:00 2001 From: Renata Date: Sat, 11 Jul 2026 00:37:16 +0200 Subject: [PATCH 1/8] feat: support choosing a YubiKey PIV slot when setting up keys Let setup-signing-key/setup-test-key target any PIV slot instead of always hardcoding SIGNATURE, add a list-slots command to see what's free/occupied, and make get-roles check every occupied slot --- taf/api/yubikey.py | 190 ++++++++++++--- taf/keys.py | 4 +- taf/tests/tuf/test_keys/test_yubikey_slots.py | 221 ++++++++++++++++++ taf/tools/yubikey/__init__.py | 83 ++++++- taf/tools/yubikey/yubikey_utils.py | 90 +++++-- taf/utils.py | 4 +- taf/yubikey/yubikey.py | 178 ++++++++++++-- 7 files changed, 685 insertions(+), 85 deletions(-) create mode 100644 taf/tests/tuf/test_keys/test_yubikey_slots.py diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index fb48e1bd2..caed96962 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -3,6 +3,7 @@ import click from pathlib import Path +from cryptography import x509 from logdecorator import log_on_end, log_on_error, log_on_start from taf.auth_repo import AuthenticationRepository from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME @@ -14,6 +15,7 @@ from taf.tuf.repository import MAIN_ROLES import taf.yubikey.yubikey as yk from taf.yubikey.yubikey_manager import PinManager +from yubikit.piv import SLOT @log_on_start(DEBUG, "Exporting public pem from YubiKey", logger=taf_logger) @@ -121,7 +123,8 @@ def export_yk_certificate( def get_yk_roles(path: str, serial: Optional[str] = None) -> Dict: """ List all roles that the inserted YubiKey whose metadata files can be signed by this YubiKey. - In case of delegated targets roles, include the delegation paths. + Every occupied PIV slot is checked. In case of delegated targets roles, include the + delegation paths. Arguments: path: Authentication repository's path. @@ -140,12 +143,15 @@ def get_yk_roles(path: str, serial: Optional[str] = None) -> Dict: auth = AuthenticationRepository(path=path) for serial in serials: - pub_key = yk.get_piv_public_key_tuf(serial=serial) - roles = auth.find_associated_roles_of_key(pub_key) - roles_with_paths: Dict = {role: {} for role in roles} - for role in roles: - if role not in MAIN_ROLES: - roles_with_paths[role] = auth.get_role_paths(role) + roles_with_paths: Dict = {} + keys = yk.get_piv_public_keys_tuf(serial=serial).get(serial, {}) + for pub_key in keys.values(): + for role in auth.find_associated_roles_of_key(pub_key): + if role in roles_with_paths: + continue + roles_with_paths[role] = ( + {} if role in MAIN_ROLES else auth.get_role_paths(role) + ) roles_per_yubikes[serial] = roles_with_paths return roles_per_yubikes @@ -160,14 +166,27 @@ def get_yk_roles(path: str, serial: Optional[str] = None) -> Dict: reraise=True, ) def setup_signing_yubikey( - pin_manager: PinManager, certs_dir: Optional[str] = None, key_size: int = 2048 + pin_manager: PinManager, + certs_dir: Optional[str] = None, + key_size: int = 2048, + slot: str = "SIGNATURE", + force: bool = False, + reset: Optional[bool] = None, ) -> None: """ - Delete everything from the inserted YubiKey, generate a new key and copy it to the YubiKey. + Generate a new key and copy it to the given PIV slot of the inserted YubiKey. Optionally export and save the certificate to a file. Arguments: certs_dir (optional): Path to a directory where the exported certificate should be stored. + slot (optional): Name of the PIV slot to set the key up in (e.g. "SIGNATURE", + "AUTHENTICATION", "KEY_MANAGEMENT", "CARD_AUTH", or "RETIRED1".."RETIRED20"). + Defaults to "SIGNATURE". + force (optional): Whether to overwrite the target slot if it already + has a key in it. Has no effect when reset is True, since resetting + always leaves every slot empty first, so there's nothing to + overwrite in that case. + reset (optional): Whether to factory-reset the card first. Defaults to False. Side Effects: None @@ -175,24 +194,101 @@ def setup_signing_yubikey( Returns: None """ - if not click.confirm( + try: + piv_slot = SLOT[slot.upper()] + except KeyError: + raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") + + resetting = bool(reset) + warning = ( "WARNING - this will delete everything from the inserted key. Proceed?" - ): - return - yubikeys = yk.yubikey_prompt( - ["new Yubikey"], - pin_manager=pin_manager, - creating_new_key=True, - pin_confirm=True, - pin_repeat=True, - prompt_message="Please insert the new Yubikey and press ENTER", + if resetting + else f"This will set up a new key in the {piv_slot.name} slot, leaving " + "the rest of the card untouched. Proceed?" ) - if yubikeys: + if not click.confirm(warning): + return + + if resetting: + # Resetting the whole card wipes its PIN too, so a new one has to be + # chosen here - this is the only case where prompting for a PIN + # makes sense. + yubikeys = yk.yubikey_prompt( + ["new Yubikey"], + pin_manager=pin_manager, + creating_new_key=True, + pin_confirm=True, + pin_repeat=True, + prompt_message="Please insert the new Yubikey and press ENTER", + ) + if not yubikeys: + raise YubikeyError("Could not generate a new key") _, serial_num, _ = yubikeys[0] - key = yk.setup_new_yubikey(pin_manager, serial_num, key_size=key_size) - yk.export_yk_certificate(certs_dir, key, serial_num) else: - raise YubikeyError("Could not generate a new key") + # Not resetting, so the card's existing PIN is left untouched and + # isn't needed for this operation (adding a key to a slot only + # requires the management key) - just find the inserted device. + input("Please insert the Yubikey and press ENTER") + serials = yk.get_serial_nums() + if not len(serials): + raise YubikeyError("YubiKey not inserted") + if len(serials) > 1: + raise YubikeyError( + "More than one YubiKey is inserted. Please insert only one" + ) + serial_num = serials[0] + + key = yk.setup_new_yubikey( + pin_manager, + serial_num, + key_size=key_size, + slot=piv_slot, + force=force, + reset=resetting, + ) + yk.export_yk_certificate(certs_dir, key, serial_num) + + +@log_on_start(DEBUG, "Listing YubiKey PIV slot status", logger=taf_logger) +@log_on_error( + ERROR, + "An error occurred while listing YubiKey PIV slots: {e}", + logger=taf_logger, + on_exceptions=TAFError, + reraise=True, +) +def list_yk_slots(serial: Optional[str] = None) -> None: + """ + Print the free/occupied status of every usable PIV slot on the inserted + YubiKey(s), including the holder name and expiry of any certificate found. + + Arguments: + serial (optional): Serial number of a specific YubiKey. Lists slots + for every inserted YubiKey if not specified. + + Side Effects: + None + + Returns: + None + """ + serials = [serial] if serial else yk.get_serial_nums() + if not len(serials): + print("YubiKey not inserted.") + return + + for dev_serial, slot_status in yk.get_slot_status(serial=serial).items(): + print(f"\nSerial: {dev_serial}") + for slot, cert in slot_status.items(): + if cert is None: + print(f" {slot.name:<15} free") + else: + cn = "" + attrs = cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME) + if attrs: + cn = attrs[0].value + expires = cert.not_valid_after_utc.strftime("%Y-%m-%d") + print(f" {slot.name:<15} occupied {cn} expires {expires}") @log_on_start(DEBUG, "Setting up a new test YubiKey", logger=taf_logger) @@ -208,13 +304,21 @@ def setup_test_yubikey( key_path: str, key_size: Optional[int] = 2048, serial: Optional[str] = None, + slot: str = "SIGNATURE", + force: bool = False, + reset: Optional[bool] = None, ) -> None: """ - Reset the inserted yubikey, set default pin and copy the specified key - to it. + Copy the specified key to the inserted YubiKey's given PIV slot. Arguments: key_path: Path to a key which should be copied to a YubiKey. + slot (optional): Name of the PIV slot to copy the key into (e.g. + "SIGNATURE", "AUTHENTICATION", "KEY_MANAGEMENT", "CARD_AUTH", or + "RETIRED1".."RETIRED20"). Defaults to "SIGNATURE". + force (optional): Whether to overwrite the target slot if it already + has a key in it. Has no effect when reset is True. + reset (optional): Whether to factory-reset the card first. Defaults to False. Side Effects: None @@ -222,6 +326,13 @@ def setup_test_yubikey( Returns: None """ + try: + piv_slot = SLOT[slot.upper()] + except KeyError: + raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") + + resetting = bool(reset) + if serial is None: serials = yk.get_serial_nums() if not len(serials): @@ -229,7 +340,13 @@ def setup_test_yubikey( if len(serials) > 1: raise YubikeyError("Insert only one YubiKey") - if not click.confirm("WARNING - this will reset the inserted key. Proceed?"): + warning = ( + "WARNING - this will reset the inserted key. Proceed?" + if resetting + else f"This will copy the key into the {piv_slot.name} slot, leaving " + "the rest of the card untouched. Proceed?" + ) + if not click.confirm(warning): return serial = serials[0] @@ -237,12 +354,25 @@ def setup_test_yubikey( key_pem = key_pem_path.read_bytes() print(f"Importing RSA private key from {key_path} to Yubikey...") - pin = yk.DEFAULT_PIN - pin_manager.add_pin(serial, pin) + if resetting: + # Resetting wipes the PIN too, so it's reset to the default here - + # for a non-reset slot add, the card's existing PIN is untouched. + pin = yk.DEFAULT_PIN + pin_manager.add_pin(serial, pin) + else: + pin = None pub_key = yk.setup( - pin, serial, "Test Yubikey", private_key_pem=key_pem, key_size=key_size + pin, + serial, + "Test Yubikey", + private_key_pem=key_pem, + key_size=key_size, + slot=piv_slot, + force=force, + reset=resetting, ) print("\nPrivate key successfully imported.\n") print("\nPublic key (PEM): \n{}".format(pub_key.decode("utf-8"))) - print("Pin: {}\n".format(pin)) + if resetting: + print("Pin: {}\n".format(pin)) diff --git a/taf/keys.py b/taf/keys.py index 5f4e177d7..6fd7cd671 100644 --- a/taf/keys.py +++ b/taf/keys.py @@ -98,8 +98,8 @@ def _get_attr(oid): "country": _get_attr(x509.OID_COUNTRY_NAME), "state": _get_attr(x509.OID_STATE_OR_PROVINCE_NAME), "locality": _get_attr(x509.OID_LOCALITY_NAME), - "valid_from": cert.not_valid_before.strftime("%Y-%m-%d"), - "valid_to": cert.not_valid_after.strftime("%Y-%m-%d"), + "valid_from": cert.not_valid_before_utc.strftime("%Y-%m-%d"), + "valid_to": cert.not_valid_after_utc.strftime("%Y-%m-%d"), } diff --git a/taf/tests/tuf/test_keys/test_yubikey_slots.py b/taf/tests/tuf/test_keys/test_yubikey_slots.py new file mode 100644 index 000000000..0cda48d79 --- /dev/null +++ b/taf/tests/tuf/test_keys/test_yubikey_slots.py @@ -0,0 +1,221 @@ +from pathlib import Path + +import pytest +from yubikit.piv import SLOT + +import taf.api.yubikey as yk_api +import taf.yubikey.yubikey as yk +from taf.exceptions import YubikeyError +from taf.tools.yubikey.yubikey_utils import FakeYubiKey, _yk_piv_ctrl_mock +from taf.yubikey.yubikey_manager import PinManager + +KEYSTORE_PATH = Path(__file__).parents[2] / "data" / "keystores" / "keystore" + + +@pytest.fixture +def fake_yubikey(monkeypatch): + monkeypatch.setattr(yk, "_yk_piv_ctrl", _yk_piv_ctrl_mock) + key = FakeYubiKey( + KEYSTORE_PATH / "root1", KEYSTORE_PATH / "root1.pub", scheme=None + ) + key.insert() + yield key + key.remove() + + +def test_get_slot_status_reports_signature_occupied_and_others_free(fake_yubikey): + status = yk.get_slot_status(serial=fake_yubikey.serial) + + slot_status = status[fake_yubikey.serial] + assert slot_status[SLOT.SIGNATURE] is not None + assert slot_status[SLOT.AUTHENTICATION] is None + assert slot_status[SLOT.KEY_MANAGEMENT] is None + + +def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey): + new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() + yk.setup( + pin="123456", + serial=fake_yubikey.serial, + cert_cn="Second key", + private_key_pem=new_key_pem, + slot=SLOT.AUTHENTICATION, + ) + + keys = yk.get_piv_public_keys_tuf(serial=fake_yubikey.serial)[ + fake_yubikey.serial + ] + + assert set(keys.keys()) == {SLOT.SIGNATURE, SLOT.AUTHENTICATION} + assert keys[SLOT.SIGNATURE].keyid == fake_yubikey.tuf_key.public_key.keyid + # the two slots must resolve to two different keys + assert keys[SLOT.SIGNATURE].keyid != keys[SLOT.AUTHENTICATION].keyid + + +def test_setup_into_free_slot_does_not_touch_existing_key(fake_yubikey): + new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() + + yk.setup( + pin="123456", + serial=fake_yubikey.serial, + cert_cn="Second key", + private_key_pem=new_key_pem, + slot=SLOT.AUTHENTICATION, + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + assert status[SLOT.AUTHENTICATION] is not None + # the original SIGNATURE key must be untouched, since we didn't reset + assert status[SLOT.SIGNATURE] is not None + original_cert = status[SLOT.SIGNATURE] + assert original_cert.public_key().public_numbers() == fake_yubikey.pub_key.public_numbers() + + +def test_setup_refuses_to_overwrite_occupied_slot_without_force(fake_yubikey): + with pytest.raises(YubikeyError): + yk.setup( + pin="123456", + serial=fake_yubikey.serial, + cert_cn="Should not be written", + slot=SLOT.SIGNATURE, + reset=False, + ) + + +def test_setup_overwrites_occupied_slot_with_force(fake_yubikey): + new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() + + yk.setup( + pin="123456", + serial=fake_yubikey.serial, + cert_cn="Replacement key", + private_key_pem=new_key_pem, + slot=SLOT.SIGNATURE, + reset=False, + force=True, + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + new_cert = status[SLOT.SIGNATURE] + assert new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() + + +def test_setup_signing_yubikey_non_reset_slot_does_not_prompt_for_new_pin( + fake_yubikey, monkeypatch +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + # avoid writing a real cert file to disk during the test + monkeypatch.setattr(yk, "export_yk_certificate", lambda *args, **kwargs: None) + + prompts = [] + + def _fake_input(prompt=""): + prompts.append(prompt) + return "Second key holder" + + monkeypatch.setattr("builtins.input", _fake_input) + + # get_pin_for is only reachable via the "creating_new_key" (reset) path; + # making it raise proves that path is never taken here. + def _fail_if_called(*args, **kwargs): + raise AssertionError("Should not prompt to set a new PIN") + + monkeypatch.setattr(yk, "get_pin_for", _fail_if_called) + + yk_api.setup_signing_yubikey( + PinManager(), key_size=2048, slot="AUTHENTICATION" + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + assert status[SLOT.AUTHENTICATION] is not None + assert status[SLOT.SIGNATURE] is not None + # only the "insert the key" prompt and the "key holder's name" prompt + # should have happened - no PIN confirm/repeat prompts + assert len(prompts) == 2 + + +def test_setup_signing_yubikey_signature_slot_with_explicit_no_reset( + fake_yubikey, monkeypatch +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + monkeypatch.setattr(yk, "export_yk_certificate", lambda *args, **kwargs: None) + monkeypatch.setattr("builtins.input", lambda prompt="": "New signer") + + def _fail_if_called(*args, **kwargs): + raise AssertionError("Should not prompt to set a new PIN") + + monkeypatch.setattr(yk, "get_pin_for", _fail_if_called) + + # SIGNATURE is already occupied by the fixture's default key, so + # force=True is required, exactly like any other non-reset overwrite. + yk_api.setup_signing_yubikey( + PinManager(), key_size=2048, slot="SIGNATURE", reset=False, force=True + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + new_cert = status[SLOT.SIGNATURE] + assert ( + new_cert.public_key().public_numbers() + != fake_yubikey.pub_key.public_numbers() + ) + + +def test_setup_test_yubikey_into_non_default_slot_does_not_reset( + fake_yubikey, monkeypatch +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + + new_key_path = KEYSTORE_PATH / "root2" + + yk_api.setup_test_yubikey( + PinManager(), str(new_key_path), slot="AUTHENTICATION" + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + assert status[SLOT.AUTHENTICATION] is not None + # the original SIGNATURE key must be untouched, since we didn't reset + assert status[SLOT.SIGNATURE] is not None + original_cert = status[SLOT.SIGNATURE] + assert ( + original_cert.public_key().public_numbers() + == fake_yubikey.pub_key.public_numbers() + ) + + +def test_setup_test_yubikey_default_does_not_reset(fake_yubikey, monkeypatch): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + + new_key_path = KEYSTORE_PATH / "root2" + + with pytest.raises(YubikeyError): + yk_api.setup_test_yubikey(PinManager(), str(new_key_path)) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + original_cert = status[SLOT.SIGNATURE] + assert ( + original_cert.public_key().public_numbers() + == fake_yubikey.pub_key.public_numbers() + ) + + +def test_setup_test_yubikey_explicit_reset_wipes_other_slots( + fake_yubikey, monkeypatch +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + + new_key_path = KEYSTORE_PATH / "root2" + + yk_api.setup_test_yubikey(PinManager(), str(new_key_path), reset=True) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + new_cert = status[SLOT.SIGNATURE] + assert ( + new_cert.public_key().public_numbers() + != fake_yubikey.pub_key.public_numbers() + ) + assert status[SLOT.AUTHENTICATION] is None diff --git a/taf/tools/yubikey/__init__.py b/taf/tools/yubikey/__init__.py index 074e9df92..c4c0cb1e7 100644 --- a/taf/tools/yubikey/__init__.py +++ b/taf/tools/yubikey/__init__.py @@ -3,6 +3,7 @@ export_yk_certificate, export_yk_public_pem, get_yk_roles, + list_yk_slots, setup_signing_yubikey, setup_test_yubikey, ) @@ -11,6 +12,9 @@ from taf.tools.cli import catch_cli_exception from taf.tools.repo import pin_managed from taf.yubikey.yubikey import list_connected_yubikeys, list_all_devices +from yubikit.piv import SLOT + +SLOT_NAMES = [s.name for s in SLOT if s != SLOT.ATTESTATION] def check_pin_command(): @@ -105,30 +109,90 @@ def list_keys(): def setup_signing_key_command(): @click.command( - help="""Generate a new key on the yubikey and set the pin. Export the generated certificate - to the specified directory. - WARNING - this will delete everything from the inserted key.""" + help="""Generate a new key on the yubikey and copy it to the given PIV slot. + Export the generated certificate to the specified directory. + WARNING - --reset will factory-reset the card, deleting everything on it first.""" ) @click.option( "--certs-dir", help="Path of the directory where the exported certificate will be saved. Set to the user home directory by default", ) + @click.option( + "--slot", + type=click.Choice(SLOT_NAMES), + default="SIGNATURE", + help="PIV slot to set the key up in. Defaults to SIGNATURE.", + ) + @click.option( + "--force", + is_flag=True, + default=False, + help="Overwrite the target slot if it's already occupied. Has no effect if " + "resetting, since that always leaves every slot empty regardless", + ) + @click.option( + "--reset/--no-reset", + default=False, + help="Whether to factory-reset the card first. Defaults to False, so only " + "the given slot is touched", + ) @catch_cli_exception(handle=YubikeyError) @pin_managed - def setup_signing_key(certs_dir, pin_manager): - setup_signing_yubikey(pin_manager, certs_dir, key_size=2048) + def setup_signing_key(certs_dir, slot, force, reset, pin_manager): + setup_signing_yubikey( + pin_manager, certs_dir, key_size=2048, slot=slot, force=force, reset=reset + ) return setup_signing_key +def list_slots_command(): + @click.command( + help="Show which PIV slots on the inserted YubiKey(s) are free and which are occupied." + ) + @click.option( + "--serial", + default=None, + help="Serial number of a specific YubiKey. Lists slots for every inserted YubiKey if not specified", + ) + @catch_cli_exception(handle=YubikeyError) + def list_slots(serial): + list_yk_slots(serial) + + return list_slots + + def setup_test_key_command(): - @click.command(help="""Copies the specified key onto the inserted YubiKey - WARNING - this will reset the inserted key.""") + @click.command( + help="""Copies the specified key onto the given PIV slot of the inserted YubiKey. + WARNING - --reset will factory-reset the card, deleting everything on it first.""" + ) @click.argument("key-path") + @click.option( + "--slot", + type=click.Choice(SLOT_NAMES), + default="SIGNATURE", + help="PIV slot to copy the key into. Defaults to SIGNATURE.", + ) + @click.option( + "--force", + is_flag=True, + default=False, + help="Overwrite the target slot if it's already occupied. Has no effect if " + "resetting, since that always leaves every slot empty regardless", + ) + @click.option( + "--reset/--no-reset", + default=False, + help="Whether to factory-reset the card first. Defaults to False, so only " + "the given slot is touched", + ) @catch_cli_exception(handle=YubikeyError) @pin_managed - def setup_test_key(key_path, pin_manager): - setup_test_yubikey(pin_manager, key_path) + def setup_test_key(key_path, slot, force, reset, pin_manager): + setup_test_yubikey( + pin_manager, key_path, slot=slot, force=force, reset=reset + ) return setup_test_key @@ -141,3 +205,4 @@ def attach_to_group(group): group.add_command(list_key_command(), name="list-key") 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_slots_command(), name="list-slots") diff --git a/taf/tools/yubikey/yubikey_utils.py b/taf/tools/yubikey/yubikey_utils.py index b43bc168f..47bf5c7c5 100644 --- a/taf/tools/yubikey/yubikey_utils.py +++ b/taf/tools/yubikey/yubikey_utils.py @@ -31,6 +31,12 @@ def __init__(self, priv_key_path, pub_key_path, scheme, serial=None, pin=VALID_P self.tuf_key = load_signer_from_file(priv_key_path) + # PIV slot state (key/cert per slot), persisted here rather than on + # FakePivController, since a new FakePivController is constructed on + # every `with _yk_piv_ctrl(...)` block - this is what needs to + # survive across separate calls (e.g. setup() then get_slot_status()) + self.slots: dict = {} + @property def driver(self): return self @@ -61,16 +67,59 @@ def remove(self): class FakePivController: + """Fake yubikit.piv.PivSession. + + Tracks key/certificate state per PIV slot, so tests can exercise + slot-targeted setup (put_key/put_certificate/get_certificate) instead of + always seeing/writing a single fixed key regardless of which slot was + requested. SLOT.SIGNATURE starts out "occupied" with the driver's own + key/cert, matching a YubiKey that's already been set up the traditional + way; every other slot starts free. + """ + def __init__(self, driver): + from yubikit.piv import SLOT + self._driver = driver + # slot state lives on the driver (FakeYubiKey), not here, so it + # survives across separate `with _yk_piv_ctrl(...)` blocks + if SLOT.SIGNATURE not in driver.slots: + driver.slots[SLOT.SIGNATURE] = { + "priv_key": driver.priv_key, + "pub_key": driver.pub_key, + "cert": self._build_certificate(driver.pub_key, driver.priv_key), + } + + @property + def _slots(self): + return self._driver.slots @property def driver(self): return None + def _build_certificate(self, pub_key, priv_key): + name = x509.Name( + [x509.NameAttribute(x509.NameOID.COMMON_NAME, self.__class__.__name__)] + ) + now = datetime.datetime.utcnow() + return ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(pub_key) + .serial_number(self._driver.serial) + .not_valid_before(now) + .not_valid_after(now + datetime.timedelta(days=365)) + .sign(priv_key, hashes.SHA256(), default_backend()) + ) + def authenticate(self, *args, **kwargs): pass + def set_management_key(self, *args, **kwargs): + pass + def change_pin(self, *args, **kwargs): pass @@ -83,36 +132,41 @@ def generate_self_signed_certificate(self, *args, **kwargs): def get_pin_tries(self): return 1 - def get_certificate(self, _slot): - name = x509.Name( - [x509.NameAttribute(x509.NameOID.COMMON_NAME, self.__class__.__name__)] - ) - now = datetime.datetime.utcnow() + def put_key(self, slot, private_key, pin_policy=None, touch_policy=None): + self._slots[slot] = { + "priv_key": private_key, + "pub_key": private_key.public_key(), + "cert": None, + } - return ( - x509.CertificateBuilder() - .subject_name(name) - .issuer_name(name) - .public_key(self._driver.pub_key) - .serial_number(self._driver.serial) - .not_valid_before(now) - .not_valid_after(now + datetime.timedelta(days=365)) - .sign(self._driver.priv_key, hashes.SHA256(), default_backend()) - ) + def put_certificate(self, slot, cert): + self._slots.setdefault(slot, {})["cert"] = cert + + def get_certificate(self, slot): + slot_data = self._slots.get(slot) + if not slot_data or slot_data.get("cert") is None: + from yubikit.core.smartcard import ApduError, SW + + raise ApduError(b"", SW.FILE_NOT_FOUND) + return slot_data["cert"] def reset(self): - pass + self._driver.slots.clear() def set_pin_retries(self, *args, **kwargs): pass + def set_pin_attempts(self, *args, **kwargs): + pass + def sign(self, slot, key_type, data, hash, padding): """Sign data using the same function as TUF""" if isinstance(data, str): data = data.encode("utf-8") - signature = self._driver.priv_key.sign(data, padding, hash) - return signature + slot_data = self._slots.get(slot) + priv_key = slot_data["priv_key"] if slot_data else self._driver.priv_key + return priv_key.sign(data, padding, hash) def verify_pin(self, pin): if self._driver.pin != pin: diff --git a/taf/utils.py b/taf/utils.py index 2b9d42add..a8af8a7d7 100644 --- a/taf/utils.py +++ b/taf/utils.py @@ -70,8 +70,8 @@ def _get_attr(oid): "country": _get_attr(x509.OID_COUNTRY_NAME), "state": _get_attr(x509.OID_STATE_OR_PROVINCE_NAME), "locality": _get_attr(x509.OID_LOCALITY_NAME), - "valid_from": cert.not_valid_before.strftime("%Y-%m-%d"), - "valid_to": cert.not_valid_after.strftime("%Y-%m-%d"), + "valid_from": cert.not_valid_before_utc.strftime("%Y-%m-%d"), + "valid_to": cert.not_valid_after_utc.strftime("%Y-%m-%d"), } diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index d2f0f811a..81dc0312e 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -25,7 +25,6 @@ MANAGEMENT_KEY_TYPE, SLOT, PivSession, - generate_random_management_key, ) from yubikit.piv import ( DEFAULT_MANAGEMENT_KEY, @@ -380,6 +379,37 @@ def get_piv_public_key_tuf( return get_sslib_key_from_value(pub_key_pem, scheme) +@raise_yubikey_err("Cannot get public keys in TUF format.") +def get_piv_public_keys_tuf( + scheme=DEFAULT_RSA_SIGNATURE_SCHEME, serial=None +) -> dict: + """Return the public key of every occupied PIV slot on a YubiKey, in + TUF's key format. + + Returns: + Dict mapping serial number to a dict of {SLOT: SSlibKey} for every + slot that currently holds a certificate. + + Raises: + - YubikeyError + """ + taf_logger.debug(f"Extracting TUF-format public keys from serial={serial}") + status = get_slot_status(serial=serial) + keys: dict = {} + for dev_serial, slot_status in status.items(): + dev_keys = {} + for slot, cert in slot_status.items(): + if cert is None: + continue + pub_key_pem = cert.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode("utf-8") + dev_keys[slot] = get_sslib_key_from_value(pub_key_pem, scheme) + keys[dev_serial] = dev_keys + return keys + + def list_connected_yubikeys(): """Lists all connected YubiKeys with their serial numbers and details.""" yubikeys = list_all_devices() @@ -621,6 +651,45 @@ def sign_piv_rsa_pkcs1v15(data, pin, serial=None): return sig +def _slot_occupied(ctrl, slot) -> bool: + """Check whether a PIV slot already has a certificate in it. + + Empty slots raise (rather than return None) when read, so absence of + an exception is what "occupied" means here. + """ + try: + ctrl.get_certificate(slot) + return True + except Exception: + return False + + +def get_slot_status(serial=None) -> dict: + """Return the certificate (or None if empty) currently in every usable + PIV slot of the inserted YubiKey(s). + + Returns: + Dict mapping serial number to a dict of {SLOT: x509.Certificate | None} + for every standard and retired slot (the attestation slot is excluded, + since it's Yubico-reserved and not usable for taf's purposes). + + Raises: + - YubikeyError + """ + usable_slots = [s for s in SLOT if s != SLOT.ATTESTATION] + with _yk_piv_ctrl(serial=serial) as sessions: + status = {} + for ctrl, dev_serial in sessions: + slot_status = {} + for slot in usable_slots: + try: + slot_status[slot] = ctrl.get_certificate(slot) + except Exception: + slot_status[slot] = None + status[dev_serial] = slot_status + return status + + @raise_yubikey_err("Cannot setup Yubikey.") def setup( pin, @@ -629,25 +698,50 @@ def setup( cert_exp_days=365, pin_retries=10, private_key_pem=None, - mgm_key=generate_random_management_key(MANAGEMENT_KEY_TYPE.TDES), + mgm_key=None, key_size=2048, + slot=SLOT.SIGNATURE, + reset=None, + management_key=None, + force=False, ): - """Use to setup inserted Yubikey, with following steps (order is important): + """Set up a key in a PIV slot of the inserted YubiKey. + + When resetting, the following steps are performed, in order: - reset to factory settings - - set management key - - generate key(RSA2048) or import given one + - authenticate with the management key + - generate key(RSA) or import given one - generate and import self-signed certificate(X509) - set pin retries - set pin - set puk(same as pin) + The management key is left at the PIV default unless mgm_key is + explicitly passed. + + When not resetting, only the given slot is touched: authenticate with + the card's existing management key, generate/import the key and its + certificate into the chosen slot, and leave every other slot, the PIN, + and the PUK untouched. Refuses to overwrite an already-occupied slot + unless force=True. + Args: - cert_cn(str): x509 common name - cert_exp_days(int): x509 expiration (in days from now) - - pin_retries(int): Number of retries for PIN + - pin_retries(int): Number of retries for PIN. Only used when resetting. - private_key_pem(str): Private key in PEM format. If given, it will be imported to Yubikey. - - mgm_key(bytes): New management key + - mgm_key(bytes): Management key to set instead of leaving it at the + PIV default. Only used when resetting. + - slot(SLOT): The PIV slot to write the key and certificate to. + Defaults to SLOT.SIGNATURE. + - reset(bool): Whether to factory-reset the card first. Defaults to + False. + - management_key(bytes): The card's current management key, used to + authenticate when reset=False. Defaults to + the PIV factory default management key. + - force(bool): When reset=False, whether to overwrite the target + slot if it's already occupied. Returns: PIV public key in PEM format (bytes) @@ -655,16 +749,37 @@ def setup( Raises: - YubikeyError """ + if reset is None: + reset = False + taf_logger.debug( - f"Initializing YubiKey setup for serial={serial}, key_size={key_size}, cert_cn='{cert_cn}'" + f"Initializing YubiKey setup for serial={serial}, key_size={key_size}, " + f"cert_cn='{cert_cn}', slot={slot}, reset={reset}" ) with _yk_piv_ctrl(serial=serial) as [(ctrl, _)]: - taf_logger.debug(f"Resetting YubiKey to factory settings for serial={serial}") - ctrl.reset() + if reset: + taf_logger.debug( + f"Resetting YubiKey to factory settings for serial={serial}" + ) + ctrl.reset() - taf_logger.debug("Setting new management key and authenticating...") - ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, DEFAULT_MANAGEMENT_KEY) - ctrl.set_management_key(MANAGEMENT_KEY_TYPE.TDES, mgm_key) + taf_logger.debug("Authenticating with the default management key...") + ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, DEFAULT_MANAGEMENT_KEY) + if mgm_key is not None: + ctrl.set_management_key(MANAGEMENT_KEY_TYPE.TDES, mgm_key) + else: + # Leave the management key at the PIV default rather than + # randomizing it - a forgotten random key would permanently + # block adding a key to any other slot without a full reset. + mgm_key = DEFAULT_MANAGEMENT_KEY + else: + if management_key is None: + management_key = DEFAULT_MANAGEMENT_KEY + ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, management_key) + if not force and _slot_occupied(ctrl, slot): + raise YubikeyError( + f"Slot {slot.name} already has a key. Pass force=True to overwrite it." + ) if private_key_pem is None: taf_logger.debug("Generating RSA private key on the fly...") @@ -684,11 +799,13 @@ def setup( private_key_pem, pem_pwd, default_backend() ) - taf_logger.debug("Placing key in SIGNATURE slot with PIN_POLICY.ALWAYS...") - ctrl.put_key(SLOT.SIGNATURE, private_key, PIN_POLICY.ALWAYS) + taf_logger.debug(f"Placing key in {slot.name} slot with PIN_POLICY.ALWAYS...") + ctrl.put_key(slot, private_key, PIN_POLICY.ALWAYS) pub_key = private_key.public_key() - ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, mgm_key) - ctrl.verify_pin(DEFAULT_PIN) + + if reset: + ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, mgm_key) + ctrl.verify_pin(DEFAULT_PIN) now = datetime.datetime.now() valid_to = now + datetime.timedelta(days=cert_exp_days) @@ -706,11 +823,13 @@ def setup( .sign(private_key, hashes.SHA256(), default_backend()) ) - ctrl.put_certificate(SLOT.SIGNATURE, cert) - taf_logger.debug("Setting PIN attempts and changing default PIN/PUK...") - ctrl.set_pin_attempts(pin_attempts=pin_retries, puk_attempts=pin_retries) - ctrl.change_pin(DEFAULT_PIN, pin) - ctrl.change_puk(DEFAULT_PUK, pin) + ctrl.put_certificate(slot, cert) + + if reset: + taf_logger.debug("Setting PIN attempts and changing default PIN/PUK...") + ctrl.set_pin_attempts(pin_attempts=pin_retries, puk_attempts=pin_retries) + ctrl.change_pin(DEFAULT_PIN, pin) + ctrl.change_puk(DEFAULT_PUK, pin) taf_logger.debug("YubiKey setup complete.") return pub_key.public_bytes( @@ -723,15 +842,26 @@ def setup_new_yubikey( serial: str, scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, key_size: Optional[int] = 2048, + slot=SLOT.SIGNATURE, + force: bool = False, + reset: Optional[bool] = None, ) -> SSlibKey: taf_logger.debug( - f"Starting new YubiKey setup for serial={serial}, scheme={scheme}, key_size={key_size}" + f"Starting new YubiKey setup for serial={serial}, scheme={scheme}, " + f"key_size={key_size}, slot={slot}, reset={reset}" ) pin = pin_manager.get_pin(serial) cert_cn = input("Enter key holder's name: ") print("Generating key, please wait...") pub_key_pem = setup( - pin, serial, cert_cn, cert_exp_days=EXPIRATION_INTERVAL, key_size=key_size + pin, + serial, + cert_cn, + cert_exp_days=EXPIRATION_INTERVAL, + key_size=key_size, + slot=slot, + force=force, + reset=reset, ).decode("utf-8") scheme = DEFAULT_RSA_SIGNATURE_SCHEME key = get_sslib_key_from_value(pub_key_pem, scheme) From b50cf5f6756b3a280c1b858c7ca0c0232380123d Mon Sep 17 00:00:00 2001 From: Renata Date: Sun, 12 Jul 2026 00:26:17 +0200 Subject: [PATCH 2/8] refactor: tighten YubiKey slot setup --- taf/api/yubikey.py | 160 +++++++++++------- taf/tests/tuf/test_keys/conftest.py | 20 +++ taf/tests/tuf/test_keys/test_yubikey_slots.py | 106 +++--------- taf/tools/yubikey/__init__.py | 18 +- taf/yubikey/yubikey.py | 16 +- 5 files changed, 163 insertions(+), 157 deletions(-) diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index caed96962..791c7a8e5 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -1,5 +1,5 @@ from logging import DEBUG, ERROR -from typing import Dict, Optional +from typing import Dict, Optional, Tuple import click from pathlib import Path @@ -17,6 +17,73 @@ from taf.yubikey.yubikey_manager import PinManager from yubikit.piv import SLOT +# Retired slots (RETIRED1-RETIRED20) are excluded here: PIV convention reserves +# them for key history/decryption, not signing, so taf doesn't offer them as a +# setup target even though the hardware itself places no such restriction. +SETUP_SLOTS = {SLOT.SIGNATURE, SLOT.AUTHENTICATION, SLOT.KEY_MANAGEMENT, SLOT.CARD_AUTH} + + +def _resolve_setup_slot(slot: str) -> SLOT: + try: + piv_slot = SLOT[slot.upper()] + except KeyError: + raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") + if piv_slot not in SETUP_SLOTS: + raise YubikeyError( + f"'{slot}' is not a supported slot for key setup. Choose one of: " + + ", ".join(s.name for s in SETUP_SLOTS) + ) + return piv_slot + + +def _resolve_single_serial(prompt: Optional[str] = None) -> str: + """Find the currently inserted YubiKey's serial number, raising if none + or more than one is inserted.""" + if prompt is not None: + input(prompt) + serials = yk.get_serial_nums() + if not len(serials): + raise YubikeyError("YubiKey not inserted") + if len(serials) > 1: + raise YubikeyError("More than one YubiKey is inserted. Please insert only one") + return serials[0] + + +def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> Optional[bool]: + """Check whether the target slot is already occupied and, if so, confirm + before overwriting it. Returns the force flag to setup with, or None if + the user declined.""" + occupied = yk.get_slot_status(serial=serial)[serial][piv_slot] is not None + if occupied and not force: + if not click.confirm( + f"WARNING - the {piv_slot.name} slot already has a key. " + "This will overwrite it. Proceed?" + ): + return None + return True + if occupied: + print(f"Overwriting the existing key in the {piv_slot.name} slot.") + else: + print(f"Setting up a new key in the {piv_slot.name} slot.") + return force + + +def _prepare_non_reset_setup( + piv_slot: SLOT, + force: bool, + serial: Optional[str] = None, + insert_prompt: Optional[str] = None, +) -> Optional[Tuple[str, bool]]: + """Resolve which YubiKey to use (unless already given) and confirm + before touching an occupied slot. Returns (serial, force) to proceed + with, or None if the user declined an overwrite confirmation.""" + if serial is None: + serial = _resolve_single_serial(insert_prompt) + force = _confirm_slot_overwrite(serial, piv_slot, force) + if force is None: + return None + return serial, force + @log_on_start(DEBUG, "Exporting public pem from YubiKey", logger=taf_logger) @log_on_end(DEBUG, "Exported public pem from YubuKey", logger=taf_logger) @@ -171,7 +238,7 @@ def setup_signing_yubikey( key_size: int = 2048, slot: str = "SIGNATURE", force: bool = False, - reset: Optional[bool] = None, + reset: bool = False, ) -> None: """ Generate a new key and copy it to the given PIV slot of the inserted YubiKey. @@ -194,25 +261,15 @@ def setup_signing_yubikey( Returns: None """ - try: - piv_slot = SLOT[slot.upper()] - except KeyError: - raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") + piv_slot = _resolve_setup_slot(slot) - resetting = bool(reset) - warning = ( - "WARNING - this will delete everything from the inserted key. Proceed?" - if resetting - else f"This will set up a new key in the {piv_slot.name} slot, leaving " - "the rest of the card untouched. Proceed?" - ) - if not click.confirm(warning): - return - - if resetting: + if reset: + if not click.confirm( + "WARNING - this will delete everything from the inserted key. Proceed?" + ): + return # Resetting the whole card wipes its PIN too, so a new one has to be - # chosen here - this is the only case where prompting for a PIN - # makes sense. + # chosen here. yubikeys = yk.yubikey_prompt( ["new Yubikey"], pin_manager=pin_manager, @@ -227,16 +284,15 @@ def setup_signing_yubikey( else: # Not resetting, so the card's existing PIN is left untouched and # isn't needed for this operation (adding a key to a slot only - # requires the management key) - just find the inserted device. - input("Please insert the Yubikey and press ENTER") - serials = yk.get_serial_nums() - if not len(serials): - raise YubikeyError("YubiKey not inserted") - if len(serials) > 1: - raise YubikeyError( - "More than one YubiKey is inserted. Please insert only one" - ) - serial_num = serials[0] + # requires the management key). + result = _prepare_non_reset_setup( + piv_slot, + force, + insert_prompt="Insert the YubiKey you want to set up and press ENTER", + ) + if result is None: + return + serial_num, force = result key = yk.setup_new_yubikey( pin_manager, @@ -244,7 +300,7 @@ def setup_signing_yubikey( key_size=key_size, slot=piv_slot, force=force, - reset=resetting, + reset=reset, ) yk.export_yk_certificate(certs_dir, key, serial_num) @@ -277,7 +333,8 @@ def list_yk_slots(serial: Optional[str] = None) -> None: print("YubiKey not inserted.") return - for dev_serial, slot_status in yk.get_slot_status(serial=serial).items(): + for dev_serial in serials: + slot_status = yk.get_slot_status(serial=dev_serial)[dev_serial] print(f"\nSerial: {dev_serial}") for slot, cert in slot_status.items(): if cert is None: @@ -306,7 +363,7 @@ def setup_test_yubikey( serial: Optional[str] = None, slot: str = "SIGNATURE", force: bool = False, - reset: Optional[bool] = None, + reset: bool = False, ) -> None: """ Copy the specified key to the inserted YubiKey's given PIV slot. @@ -326,35 +383,24 @@ def setup_test_yubikey( Returns: None """ - try: - piv_slot = SLOT[slot.upper()] - except KeyError: - raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") - - resetting = bool(reset) + piv_slot = _resolve_setup_slot(slot) - if serial is None: - serials = yk.get_serial_nums() - if not len(serials): - raise YubikeyError("YubiKey not inserted") - if len(serials) > 1: - raise YubikeyError("Insert only one YubiKey") - - warning = ( - "WARNING - this will reset the inserted key. Proceed?" - if resetting - else f"This will copy the key into the {piv_slot.name} slot, leaving " - "the rest of the card untouched. Proceed?" - ) - if not click.confirm(warning): - return + if reset: + if serial is None: + serial = _resolve_single_serial() + if not click.confirm("WARNING - this will reset the inserted key. Proceed?"): + return + else: + result = _prepare_non_reset_setup(piv_slot, force, serial=serial) + if result is None: + return + serial, force = result - serial = serials[0] key_pem_path = Path(key_path) key_pem = key_pem_path.read_bytes() print(f"Importing RSA private key from {key_path} to Yubikey...") - if resetting: + if reset: # Resetting wipes the PIN too, so it's reset to the default here - # for a non-reset slot add, the card's existing PIN is untouched. pin = yk.DEFAULT_PIN @@ -370,9 +416,9 @@ def setup_test_yubikey( key_size=key_size, slot=piv_slot, force=force, - reset=resetting, + reset=reset, ) print("\nPrivate key successfully imported.\n") print("\nPublic key (PEM): \n{}".format(pub_key.decode("utf-8"))) - if resetting: + if reset: print("Pin: {}\n".format(pin)) diff --git a/taf/tests/tuf/test_keys/conftest.py b/taf/tests/tuf/test_keys/conftest.py index e3c9723e2..f4d271ec6 100644 --- a/taf/tests/tuf/test_keys/conftest.py +++ b/taf/tests/tuf/test_keys/conftest.py @@ -1,3 +1,5 @@ +from pathlib import Path + import pytest import shutil @@ -6,6 +8,8 @@ from taf.models.types import RolesKeysData from taf.tuf.repository import MetadataRepository +KEYSTORE_PATH = Path(__file__).parents[2] / "data" / "keystores" / "keystore" + @pytest.fixture def fake_single_yubikey(monkeypatch): @@ -18,6 +22,22 @@ def fake_single_yubikey(monkeypatch): monkeypatch.setattr(yk, "get_serial_nums", lambda: ["1234"]) +@pytest.fixture +def fake_yubikey(monkeypatch): + """A fake inserted YubiKey backed by a real keystore key in SIGNATURE, + with put_key/put_certificate/reset emulated per-slot so slot-management + behavior (setup, get_slot_status, get_piv_public_keys_tuf) can be tested + without real hardware.""" + import taf.yubikey.yubikey as yk + from taf.tools.yubikey.yubikey_utils import FakeYubiKey, _yk_piv_ctrl_mock + + monkeypatch.setattr(yk, "_yk_piv_ctrl", _yk_piv_ctrl_mock) + key = FakeYubiKey(KEYSTORE_PATH / "root1", KEYSTORE_PATH / "root1.pub", scheme=None) + key.insert() + yield key + key.remove() + + @pytest.fixture(autouse=False) def tuf_repo( tuf_repo_path, signers_with_delegations, with_delegations_no_yubikeys_input diff --git a/taf/tests/tuf/test_keys/test_yubikey_slots.py b/taf/tests/tuf/test_keys/test_yubikey_slots.py index 0cda48d79..5cd7be45b 100644 --- a/taf/tests/tuf/test_keys/test_yubikey_slots.py +++ b/taf/tests/tuf/test_keys/test_yubikey_slots.py @@ -6,21 +6,14 @@ import taf.api.yubikey as yk_api import taf.yubikey.yubikey as yk from taf.exceptions import YubikeyError -from taf.tools.yubikey.yubikey_utils import FakeYubiKey, _yk_piv_ctrl_mock from taf.yubikey.yubikey_manager import PinManager KEYSTORE_PATH = Path(__file__).parents[2] / "data" / "keystores" / "keystore" -@pytest.fixture -def fake_yubikey(monkeypatch): - monkeypatch.setattr(yk, "_yk_piv_ctrl", _yk_piv_ctrl_mock) - key = FakeYubiKey( - KEYSTORE_PATH / "root1", KEYSTORE_PATH / "root1.pub", scheme=None - ) - key.insert() - yield key - key.remove() +def test_resolve_setup_slot_rejects_retired_slots(): + with pytest.raises(YubikeyError): + yk_api._resolve_setup_slot("RETIRED1") def test_get_slot_status_reports_signature_occupied_and_others_free(fake_yubikey): @@ -42,9 +35,7 @@ def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey): slot=SLOT.AUTHENTICATION, ) - keys = yk.get_piv_public_keys_tuf(serial=fake_yubikey.serial)[ - fake_yubikey.serial - ] + keys = yk.get_piv_public_keys_tuf(serial=fake_yubikey.serial)[fake_yubikey.serial] assert set(keys.keys()) == {SLOT.SIGNATURE, SLOT.AUTHENTICATION} assert keys[SLOT.SIGNATURE].keyid == fake_yubikey.tuf_key.public_key.keyid @@ -68,7 +59,10 @@ def test_setup_into_free_slot_does_not_touch_existing_key(fake_yubikey): # the original SIGNATURE key must be untouched, since we didn't reset assert status[SLOT.SIGNATURE] is not None original_cert = status[SLOT.SIGNATURE] - assert original_cert.public_key().public_numbers() == fake_yubikey.pub_key.public_numbers() + assert ( + original_cert.public_key().public_numbers() + == fake_yubikey.pub_key.public_numbers() + ) def test_setup_refuses_to_overwrite_occupied_slot_without_force(fake_yubikey): @@ -97,57 +91,18 @@ def test_setup_overwrites_occupied_slot_with_force(fake_yubikey): status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] new_cert = status[SLOT.SIGNATURE] - assert new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() - - -def test_setup_signing_yubikey_non_reset_slot_does_not_prompt_for_new_pin( - fake_yubikey, monkeypatch -): - monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) - # avoid writing a real cert file to disk during the test - monkeypatch.setattr(yk, "export_yk_certificate", lambda *args, **kwargs: None) - - prompts = [] - - def _fake_input(prompt=""): - prompts.append(prompt) - return "Second key holder" - - monkeypatch.setattr("builtins.input", _fake_input) - - # get_pin_for is only reachable via the "creating_new_key" (reset) path; - # making it raise proves that path is never taken here. - def _fail_if_called(*args, **kwargs): - raise AssertionError("Should not prompt to set a new PIN") - - monkeypatch.setattr(yk, "get_pin_for", _fail_if_called) - - yk_api.setup_signing_yubikey( - PinManager(), key_size=2048, slot="AUTHENTICATION" + assert ( + new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() ) - status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] - assert status[SLOT.AUTHENTICATION] is not None - assert status[SLOT.SIGNATURE] is not None - # only the "insert the key" prompt and the "key holder's name" prompt - # should have happened - no PIN confirm/repeat prompts - assert len(prompts) == 2 - -def test_setup_signing_yubikey_signature_slot_with_explicit_no_reset( +def test_setup_signing_yubikey_can_write_signature_slot_without_resetting( fake_yubikey, monkeypatch ): monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) monkeypatch.setattr(yk, "export_yk_certificate", lambda *args, **kwargs: None) monkeypatch.setattr("builtins.input", lambda prompt="": "New signer") - def _fail_if_called(*args, **kwargs): - raise AssertionError("Should not prompt to set a new PIN") - - monkeypatch.setattr(yk, "get_pin_for", _fail_if_called) - # SIGNATURE is already occupied by the fixture's default key, so # force=True is required, exactly like any other non-reset overwrite. yk_api.setup_signing_yubikey( @@ -157,27 +112,20 @@ def _fail_if_called(*args, **kwargs): status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] new_cert = status[SLOT.SIGNATURE] assert ( - new_cert.public_key().public_numbers() - != fake_yubikey.pub_key.public_numbers() + new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() ) -def test_setup_test_yubikey_into_non_default_slot_does_not_reset( +def test_setup_test_yubikey_declining_occupied_slot_confirmation_leaves_key_untouched( fake_yubikey, monkeypatch ): - monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: False) new_key_path = KEYSTORE_PATH / "root2" - yk_api.setup_test_yubikey( - PinManager(), str(new_key_path), slot="AUTHENTICATION" - ) + yk_api.setup_test_yubikey(PinManager(), str(new_key_path)) status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] - assert status[SLOT.AUTHENTICATION] is not None - # the original SIGNATURE key must be untouched, since we didn't reset - assert status[SLOT.SIGNATURE] is not None original_cert = status[SLOT.SIGNATURE] assert ( original_cert.public_key().public_numbers() @@ -185,28 +133,23 @@ def test_setup_test_yubikey_into_non_default_slot_does_not_reset( ) -def test_setup_test_yubikey_default_does_not_reset(fake_yubikey, monkeypatch): - monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) - +def test_setup_test_yubikey_force_skips_occupied_slot_confirmation(fake_yubikey): + # force=True on an already-occupied slot must succeed without any + # interactive confirmation, so click.confirm is deliberately left + # unmocked here. new_key_path = KEYSTORE_PATH / "root2" - with pytest.raises(YubikeyError): - yk_api.setup_test_yubikey(PinManager(), str(new_key_path)) + yk_api.setup_test_yubikey(PinManager(), str(new_key_path), force=True) status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] - original_cert = status[SLOT.SIGNATURE] + new_cert = status[SLOT.SIGNATURE] assert ( - original_cert.public_key().public_numbers() - == fake_yubikey.pub_key.public_numbers() + new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() ) -def test_setup_test_yubikey_explicit_reset_wipes_other_slots( - fake_yubikey, monkeypatch -): +def test_setup_test_yubikey_explicit_reset_wipes_other_slots(fake_yubikey, monkeypatch): monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - monkeypatch.setattr(yk, "get_serial_nums", lambda: [fake_yubikey.serial]) new_key_path = KEYSTORE_PATH / "root2" @@ -215,7 +158,6 @@ def test_setup_test_yubikey_explicit_reset_wipes_other_slots( status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] new_cert = status[SLOT.SIGNATURE] assert ( - new_cert.public_key().public_numbers() - != fake_yubikey.pub_key.public_numbers() + new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() ) assert status[SLOT.AUTHENTICATION] is None diff --git a/taf/tools/yubikey/__init__.py b/taf/tools/yubikey/__init__.py index c4c0cb1e7..9d6eb0c95 100644 --- a/taf/tools/yubikey/__init__.py +++ b/taf/tools/yubikey/__init__.py @@ -1,5 +1,6 @@ import click from taf.api.yubikey import ( + SETUP_SLOTS, export_yk_certificate, export_yk_public_pem, get_yk_roles, @@ -12,9 +13,8 @@ from taf.tools.cli import catch_cli_exception from taf.tools.repo import pin_managed from taf.yubikey.yubikey import list_connected_yubikeys, list_all_devices -from yubikit.piv import SLOT -SLOT_NAMES = [s.name for s in SLOT if s != SLOT.ATTESTATION] +SETUP_SLOT_NAMES = [s.name for s in SETUP_SLOTS] def check_pin_command(): @@ -119,7 +119,7 @@ def setup_signing_key_command(): ) @click.option( "--slot", - type=click.Choice(SLOT_NAMES), + type=click.Choice(SETUP_SLOT_NAMES), default="SIGNATURE", help="PIV slot to set the key up in. Defaults to SIGNATURE.", ) @@ -133,8 +133,7 @@ def setup_signing_key_command(): @click.option( "--reset/--no-reset", default=False, - help="Whether to factory-reset the card first. Defaults to False, so only " - "the given slot is touched", + help="Whether to factory-reset the card first. Defaults to False.", ) @catch_cli_exception(handle=YubikeyError) @pin_managed @@ -170,7 +169,7 @@ def setup_test_key_command(): @click.argument("key-path") @click.option( "--slot", - type=click.Choice(SLOT_NAMES), + type=click.Choice(SETUP_SLOT_NAMES), default="SIGNATURE", help="PIV slot to copy the key into. Defaults to SIGNATURE.", ) @@ -184,15 +183,12 @@ def setup_test_key_command(): @click.option( "--reset/--no-reset", default=False, - help="Whether to factory-reset the card first. Defaults to False, so only " - "the given slot is touched", + help="Whether to factory-reset the card first. Defaults to False.", ) @catch_cli_exception(handle=YubikeyError) @pin_managed def setup_test_key(key_path, slot, force, reset, pin_manager): - setup_test_yubikey( - pin_manager, key_path, slot=slot, force=force, reset=reset - ) + setup_test_yubikey(pin_manager, key_path, slot=slot, force=force, reset=reset) return setup_test_key diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index 81dc0312e..57dda5bf9 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -380,9 +380,7 @@ def get_piv_public_key_tuf( @raise_yubikey_err("Cannot get public keys in TUF format.") -def get_piv_public_keys_tuf( - scheme=DEFAULT_RSA_SIGNATURE_SCHEME, serial=None -) -> dict: +def get_piv_public_keys_tuf(scheme=DEFAULT_RSA_SIGNATURE_SCHEME, serial=None) -> dict: """Return the public key of every occupied PIV slot on a YubiKey, in TUF's key format. @@ -401,10 +399,14 @@ def get_piv_public_keys_tuf( for slot, cert in slot_status.items(): if cert is None: continue - pub_key_pem = cert.public_key().public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode("utf-8") + pub_key_pem = ( + cert.public_key() + .public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode("utf-8") + ) dev_keys[slot] = get_sslib_key_from_value(pub_key_pem, scheme) keys[dev_serial] = dev_keys return keys From 3e6e10eb685e4d5f81b2f8164b0c4de91af896b3 Mon Sep 17 00:00:00 2001 From: Renata Date: Sun, 12 Jul 2026 01:11:41 +0200 Subject: [PATCH 3/8] refactor: simplify slot-overwrite confirmation helpers --- taf/api/yubikey.py | 38 ++++++++++++++++++-------------------- 1 file changed, 18 insertions(+), 20 deletions(-) diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index 791c7a8e5..821fd5596 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -49,23 +49,21 @@ def _resolve_single_serial(prompt: Optional[str] = None) -> str: return serials[0] -def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> Optional[bool]: +def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> bool: """Check whether the target slot is already occupied and, if so, confirm - before overwriting it. Returns the force flag to setup with, or None if - the user declined.""" + before overwriting it. Returns whether to proceed.""" occupied = yk.get_slot_status(serial=serial)[serial][piv_slot] is not None if occupied and not force: if not click.confirm( f"WARNING - the {piv_slot.name} slot already has a key. " "This will overwrite it. Proceed?" ): - return None - return True - if occupied: + return False + elif occupied: print(f"Overwriting the existing key in the {piv_slot.name} slot.") else: print(f"Setting up a new key in the {piv_slot.name} slot.") - return force + return True def _prepare_non_reset_setup( @@ -73,16 +71,16 @@ def _prepare_non_reset_setup( force: bool, serial: Optional[str] = None, insert_prompt: Optional[str] = None, -) -> Optional[Tuple[str, bool]]: +) -> Tuple[bool, str]: """Resolve which YubiKey to use (unless already given) and confirm - before touching an occupied slot. Returns (serial, force) to proceed - with, or None if the user declined an overwrite confirmation.""" + before touching an occupied slot. Returns (proceed, serial); once + proceed is True, the target slot's occupancy has already been dealt + with, so it's safe to setup with force=True regardless of the force + that was passed in.""" if serial is None: serial = _resolve_single_serial(insert_prompt) - force = _confirm_slot_overwrite(serial, piv_slot, force) - if force is None: - return None - return serial, force + proceed = _confirm_slot_overwrite(serial, piv_slot, force) + return proceed, serial @log_on_start(DEBUG, "Exporting public pem from YubiKey", logger=taf_logger) @@ -285,14 +283,14 @@ def setup_signing_yubikey( # Not resetting, so the card's existing PIN is left untouched and # isn't needed for this operation (adding a key to a slot only # requires the management key). - result = _prepare_non_reset_setup( + proceed, serial_num = _prepare_non_reset_setup( piv_slot, force, insert_prompt="Insert the YubiKey you want to set up and press ENTER", ) - if result is None: + if not proceed: return - serial_num, force = result + force = True key = yk.setup_new_yubikey( pin_manager, @@ -391,10 +389,10 @@ def setup_test_yubikey( if not click.confirm("WARNING - this will reset the inserted key. Proceed?"): return else: - result = _prepare_non_reset_setup(piv_slot, force, serial=serial) - if result is None: + proceed, serial = _prepare_non_reset_setup(piv_slot, force, serial=serial) + if not proceed: return - serial, force = result + force = True key_pem_path = Path(key_path) key_pem = key_pem_path.read_bytes() From 9c0af4b7638e1df83a4e0003f8e351d104636304 Mon Sep 17 00:00:00 2001 From: Renata Date: Mon, 13 Jul 2026 22:34:53 +0200 Subject: [PATCH 4/8] fix: detect PIV slot occupancy by key presence, not certificate _slot_occupied only checked for a certificate, missing a slot that holds a key without one (e.g. an interrupted setup(), or a slot provisioned by another tool) - such a slot would be silently overwritten even without --force. Now checks get_slot_metadata first (reflects the key directly), falling back to the certificate check on firmware that doesn't support it (pre-5.3, e.g. YubiKey 4). --- taf/api/yubikey.py | 168 +++++++++--------- taf/tests/tuf/test_keys/conftest.py | 18 +- taf/tests/tuf/test_keys/test_yubikey_slots.py | 71 +++++--- taf/tools/yubikey/__init__.py | 32 ++-- taf/tools/yubikey/yubikey_utils.py | 16 +- taf/yubikey/yubikey.py | 36 +++- 6 files changed, 190 insertions(+), 151 deletions(-) diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index 821fd5596..49be04d62 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -17,42 +17,14 @@ from taf.yubikey.yubikey_manager import PinManager from yubikit.piv import SLOT -# Retired slots (RETIRED1-RETIRED20) are excluded here: PIV convention reserves -# them for key history/decryption, not signing, so taf doesn't offer them as a -# setup target even though the hardware itself places no such restriction. +# Retired slots (RETIRED1-RETIRED20) are reserved for key history/decryption, not signing. SETUP_SLOTS = {SLOT.SIGNATURE, SLOT.AUTHENTICATION, SLOT.KEY_MANAGEMENT, SLOT.CARD_AUTH} -def _resolve_setup_slot(slot: str) -> SLOT: - try: - piv_slot = SLOT[slot.upper()] - except KeyError: - raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") - if piv_slot not in SETUP_SLOTS: - raise YubikeyError( - f"'{slot}' is not a supported slot for key setup. Choose one of: " - + ", ".join(s.name for s in SETUP_SLOTS) - ) - return piv_slot - - -def _resolve_single_serial(prompt: Optional[str] = None) -> str: - """Find the currently inserted YubiKey's serial number, raising if none - or more than one is inserted.""" - if prompt is not None: - input(prompt) - serials = yk.get_serial_nums() - if not len(serials): - raise YubikeyError("YubiKey not inserted") - if len(serials) > 1: - raise YubikeyError("More than one YubiKey is inserted. Please insert only one") - return serials[0] - - def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> bool: """Check whether the target slot is already occupied and, if so, confirm before overwriting it. Returns whether to proceed.""" - occupied = yk.get_slot_status(serial=serial)[serial][piv_slot] is not None + occupied = yk.is_slot_occupied(serial, piv_slot) if occupied and not force: if not click.confirm( f"WARNING - the {piv_slot.name} slot already has a key. " @@ -83,6 +55,32 @@ def _prepare_non_reset_setup( return proceed, serial +def _resolve_single_serial(prompt: Optional[str] = None) -> str: + """Find the currently inserted YubiKey's serial number, raising if none + or more than one is inserted.""" + if prompt is not None: + input(prompt) + serials = yk.get_serial_nums() + if not len(serials): + raise YubikeyError("YubiKey not inserted") + if len(serials) > 1: + raise YubikeyError("More than one YubiKey is inserted. Please insert only one") + return serials[0] + + +def _resolve_slot(slot: str) -> SLOT: + try: + piv_slot = SLOT[slot.upper()] + except KeyError: + raise YubikeyError(f"'{slot}' is not a valid YubiKey PIV slot name") + if piv_slot not in SETUP_SLOTS: + raise YubikeyError( + f"'{slot}' is not a supported slot for key setup. Choose one of: " + + ", ".join(s.name for s in SETUP_SLOTS) + ) + return piv_slot + + @log_on_start(DEBUG, "Exporting public pem from YubiKey", logger=taf_logger) @log_on_end(DEBUG, "Exported public pem from YubuKey", logger=taf_logger) @log_on_error( @@ -221,6 +219,49 @@ def get_yk_roles(path: str, serial: Optional[str] = None) -> Dict: return roles_per_yubikes +@log_on_start(DEBUG, "Listing YubiKey PIV slot status", logger=taf_logger) +@log_on_error( + ERROR, + "An error occurred while listing YubiKey PIV slots: {e}", + logger=taf_logger, + on_exceptions=TAFError, + reraise=True, +) +def list_yk_slots(serial: Optional[str] = None) -> None: + """ + Print the free/occupied status of every usable PIV slot on the inserted + YubiKey(s), including the holder name and expiry of any certificate found. + + Arguments: + serial (optional): Serial number of a specific YubiKey. Lists slots + for every inserted YubiKey if not specified. + + Side Effects: + None + + Returns: + None + """ + serials = [serial] if serial else yk.get_serial_nums() + if not len(serials): + print("YubiKey not inserted.") + return + + for dev_serial in serials: + slot_status = yk.get_slot_status(serial=dev_serial)[dev_serial] + print(f"\nSerial: {dev_serial}") + for slot, cert in slot_status.items(): + if cert is None: + print(f" {slot.name:<15} free") + else: + cn = "" + attrs = cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME) + if attrs: + cn = attrs[0].value + expires = cert.not_valid_after_utc.strftime("%Y-%m-%d") + print(f" {slot.name:<15} occupied {cn} expires {expires}") + + @log_on_start(DEBUG, "Setting up a new signing YubiKey", logger=taf_logger) @log_on_end(DEBUG, "Finished setting up a new signing YubiKey", logger=taf_logger) @log_on_error( @@ -244,9 +285,8 @@ def setup_signing_yubikey( Arguments: certs_dir (optional): Path to a directory where the exported certificate should be stored. - slot (optional): Name of the PIV slot to set the key up in (e.g. "SIGNATURE", - "AUTHENTICATION", "KEY_MANAGEMENT", "CARD_AUTH", or "RETIRED1".."RETIRED20"). - Defaults to "SIGNATURE". + slot (optional): Name of the PIV slot to set the key up in ("SIGNATURE", + "AUTHENTICATION", "KEY_MANAGEMENT", or "CARD_AUTH"). Defaults to "SIGNATURE". force (optional): Whether to overwrite the target slot if it already has a key in it. Has no effect when reset is True, since resetting always leaves every slot empty first, so there's nothing to @@ -259,15 +299,14 @@ def setup_signing_yubikey( Returns: None """ - piv_slot = _resolve_setup_slot(slot) + piv_slot = _resolve_slot(slot) if reset: if not click.confirm( "WARNING - this will delete everything from the inserted key. Proceed?" ): return - # Resetting the whole card wipes its PIN too, so a new one has to be - # chosen here. + # resetting wipes the PIN too, so a new one is chosen here yubikeys = yk.yubikey_prompt( ["new Yubikey"], pin_manager=pin_manager, @@ -280,9 +319,7 @@ def setup_signing_yubikey( raise YubikeyError("Could not generate a new key") _, serial_num, _ = yubikeys[0] else: - # Not resetting, so the card's existing PIN is left untouched and - # isn't needed for this operation (adding a key to a slot only - # requires the management key). + # not resetting, so the PIN is untouched and not needed here proceed, serial_num = _prepare_non_reset_setup( piv_slot, force, @@ -303,49 +340,6 @@ def setup_signing_yubikey( yk.export_yk_certificate(certs_dir, key, serial_num) -@log_on_start(DEBUG, "Listing YubiKey PIV slot status", logger=taf_logger) -@log_on_error( - ERROR, - "An error occurred while listing YubiKey PIV slots: {e}", - logger=taf_logger, - on_exceptions=TAFError, - reraise=True, -) -def list_yk_slots(serial: Optional[str] = None) -> None: - """ - Print the free/occupied status of every usable PIV slot on the inserted - YubiKey(s), including the holder name and expiry of any certificate found. - - Arguments: - serial (optional): Serial number of a specific YubiKey. Lists slots - for every inserted YubiKey if not specified. - - Side Effects: - None - - Returns: - None - """ - serials = [serial] if serial else yk.get_serial_nums() - if not len(serials): - print("YubiKey not inserted.") - return - - for dev_serial in serials: - slot_status = yk.get_slot_status(serial=dev_serial)[dev_serial] - print(f"\nSerial: {dev_serial}") - for slot, cert in slot_status.items(): - if cert is None: - print(f" {slot.name:<15} free") - else: - cn = "" - attrs = cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME) - if attrs: - cn = attrs[0].value - expires = cert.not_valid_after_utc.strftime("%Y-%m-%d") - print(f" {slot.name:<15} occupied {cn} expires {expires}") - - @log_on_start(DEBUG, "Setting up a new test YubiKey", logger=taf_logger) @log_on_end(DEBUG, "Finished setting up a test YubiKey", logger=taf_logger) @log_on_error( @@ -368,9 +362,8 @@ def setup_test_yubikey( Arguments: key_path: Path to a key which should be copied to a YubiKey. - slot (optional): Name of the PIV slot to copy the key into (e.g. - "SIGNATURE", "AUTHENTICATION", "KEY_MANAGEMENT", "CARD_AUTH", or - "RETIRED1".."RETIRED20"). Defaults to "SIGNATURE". + slot (optional): Name of the PIV slot to copy the key into ("SIGNATURE", + "AUTHENTICATION", "KEY_MANAGEMENT", or "CARD_AUTH"). Defaults to "SIGNATURE". force (optional): Whether to overwrite the target slot if it already has a key in it. Has no effect when reset is True. reset (optional): Whether to factory-reset the card first. Defaults to False. @@ -381,7 +374,7 @@ def setup_test_yubikey( Returns: None """ - piv_slot = _resolve_setup_slot(slot) + piv_slot = _resolve_slot(slot) if reset: if serial is None: @@ -399,8 +392,7 @@ def setup_test_yubikey( print(f"Importing RSA private key from {key_path} to Yubikey...") if reset: - # Resetting wipes the PIN too, so it's reset to the default here - - # for a non-reset slot add, the card's existing PIN is untouched. + # resetting wipes the PIN too, so it's reset to the default here pin = yk.DEFAULT_PIN pin_manager.add_pin(serial, pin) else: diff --git a/taf/tests/tuf/test_keys/conftest.py b/taf/tests/tuf/test_keys/conftest.py index f4d271ec6..26067047c 100644 --- a/taf/tests/tuf/test_keys/conftest.py +++ b/taf/tests/tuf/test_keys/conftest.py @@ -1,38 +1,28 @@ -from pathlib import Path - import pytest import shutil +import taf.yubikey.yubikey as yk +from taf.tools.yubikey.yubikey_utils import FakeYubiKey, _yk_piv_ctrl_mock from taf.utils import on_rm_error from taf.models.converter import from_dict from taf.models.types import RolesKeysData from taf.tuf.repository import MetadataRepository -KEYSTORE_PATH = Path(__file__).parents[2] / "data" / "keystores" / "keystore" - @pytest.fixture def fake_single_yubikey(monkeypatch): """Pretend exactly one YubiKey (serial ``1234``) is inserted.""" - # Imported lazily: taf.yubikey.yubikey pulls in the optional `ykman` - # dependency, which need not be installed to collect the rest of this - # directory's tests. - import taf.yubikey.yubikey as yk - monkeypatch.setattr(yk, "get_serial_nums", lambda: ["1234"]) @pytest.fixture -def fake_yubikey(monkeypatch): +def fake_yubikey(monkeypatch, keystore): """A fake inserted YubiKey backed by a real keystore key in SIGNATURE, with put_key/put_certificate/reset emulated per-slot so slot-management behavior (setup, get_slot_status, get_piv_public_keys_tuf) can be tested without real hardware.""" - import taf.yubikey.yubikey as yk - from taf.tools.yubikey.yubikey_utils import FakeYubiKey, _yk_piv_ctrl_mock - monkeypatch.setattr(yk, "_yk_piv_ctrl", _yk_piv_ctrl_mock) - key = FakeYubiKey(KEYSTORE_PATH / "root1", KEYSTORE_PATH / "root1.pub", scheme=None) + key = FakeYubiKey(keystore / "root1", keystore / "root1.pub", scheme=None) key.insert() yield key key.remove() diff --git a/taf/tests/tuf/test_keys/test_yubikey_slots.py b/taf/tests/tuf/test_keys/test_yubikey_slots.py index 5cd7be45b..fb1ddf333 100644 --- a/taf/tests/tuf/test_keys/test_yubikey_slots.py +++ b/taf/tests/tuf/test_keys/test_yubikey_slots.py @@ -1,32 +1,60 @@ -from pathlib import Path - import pytest +from yubikit.core import NotSupportedError from yubikit.piv import SLOT import taf.api.yubikey as yk_api import taf.yubikey.yubikey as yk from taf.exceptions import YubikeyError +from taf.tools.yubikey.yubikey_utils import FakePivController from taf.yubikey.yubikey_manager import PinManager -KEYSTORE_PATH = Path(__file__).parents[2] / "data" / "keystores" / "keystore" - -def test_resolve_setup_slot_rejects_retired_slots(): +def test_resolve_slot_rejects_retired_slots(): with pytest.raises(YubikeyError): - yk_api._resolve_setup_slot("RETIRED1") + yk_api._resolve_slot("RETIRED1") def test_get_slot_status_reports_signature_occupied_and_others_free(fake_yubikey): status = yk.get_slot_status(serial=fake_yubikey.serial) slot_status = status[fake_yubikey.serial] - assert slot_status[SLOT.SIGNATURE] is not None + assert ( + slot_status[SLOT.SIGNATURE].public_key().public_numbers() + == fake_yubikey.pub_key.public_numbers() + ) assert slot_status[SLOT.AUTHENTICATION] is None assert slot_status[SLOT.KEY_MANAGEMENT] is None -def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey): - new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() +def test_is_slot_occupied_detects_key_without_certificate(fake_yubikey): + # a slot's key and certificate are separate objects - simulate one + # holding a key but no certificate (e.g. an interrupted setup(), or a + # slot provisioned by another tool) and confirm occupancy is still + # detected, rather than relying on certificate presence alone + fake_yubikey.slots[SLOT.AUTHENTICATION] = { + "priv_key": fake_yubikey.priv_key, + "pub_key": fake_yubikey.pub_key, + "cert": None, + } + + assert yk.is_slot_occupied(fake_yubikey.serial, SLOT.AUTHENTICATION) + + +def test_is_slot_occupied_falls_back_to_certificate_on_older_firmware( + fake_yubikey, monkeypatch +): + def _unsupported(self, slot): + raise NotSupportedError("get_slot_metadata requires firmware 5.3+") + + monkeypatch.setattr(FakePivController, "get_slot_metadata", _unsupported) + + # SIGNATURE has both a key and a certificate (the fixture's default), so + # the certificate-based fallback must still detect it as occupied + assert yk.is_slot_occupied(fake_yubikey.serial, SLOT.SIGNATURE) + + +def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey, keystore): + new_key_pem = (keystore / "root2").read_bytes() yk.setup( pin="123456", serial=fake_yubikey.serial, @@ -43,8 +71,8 @@ def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey): assert keys[SLOT.SIGNATURE].keyid != keys[SLOT.AUTHENTICATION].keyid -def test_setup_into_free_slot_does_not_touch_existing_key(fake_yubikey): - new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() +def test_setup_into_free_slot_does_not_touch_existing_key(fake_yubikey, keystore): + new_key_pem = (keystore / "root2").read_bytes() yk.setup( pin="123456", @@ -57,7 +85,6 @@ def test_setup_into_free_slot_does_not_touch_existing_key(fake_yubikey): status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] assert status[SLOT.AUTHENTICATION] is not None # the original SIGNATURE key must be untouched, since we didn't reset - assert status[SLOT.SIGNATURE] is not None original_cert = status[SLOT.SIGNATURE] assert ( original_cert.public_key().public_numbers() @@ -76,8 +103,8 @@ def test_setup_refuses_to_overwrite_occupied_slot_without_force(fake_yubikey): ) -def test_setup_overwrites_occupied_slot_with_force(fake_yubikey): - new_key_pem = (KEYSTORE_PATH / "root2").read_bytes() +def test_setup_overwrites_occupied_slot_with_force(fake_yubikey, keystore): + new_key_pem = (keystore / "root2").read_bytes() yk.setup( pin="123456", @@ -117,11 +144,11 @@ def test_setup_signing_yubikey_can_write_signature_slot_without_resetting( def test_setup_test_yubikey_declining_occupied_slot_confirmation_leaves_key_untouched( - fake_yubikey, monkeypatch + fake_yubikey, monkeypatch, keystore ): monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: False) - new_key_path = KEYSTORE_PATH / "root2" + new_key_path = keystore / "root2" yk_api.setup_test_yubikey(PinManager(), str(new_key_path)) @@ -133,11 +160,13 @@ def test_setup_test_yubikey_declining_occupied_slot_confirmation_leaves_key_unto ) -def test_setup_test_yubikey_force_skips_occupied_slot_confirmation(fake_yubikey): +def test_setup_test_yubikey_force_skips_occupied_slot_confirmation( + fake_yubikey, keystore +): # force=True on an already-occupied slot must succeed without any # interactive confirmation, so click.confirm is deliberately left # unmocked here. - new_key_path = KEYSTORE_PATH / "root2" + new_key_path = keystore / "root2" yk_api.setup_test_yubikey(PinManager(), str(new_key_path), force=True) @@ -148,10 +177,12 @@ def test_setup_test_yubikey_force_skips_occupied_slot_confirmation(fake_yubikey) ) -def test_setup_test_yubikey_explicit_reset_wipes_other_slots(fake_yubikey, monkeypatch): +def test_setup_test_yubikey_explicit_reset_wipes_other_slots( + fake_yubikey, monkeypatch, keystore +): monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) - new_key_path = KEYSTORE_PATH / "root2" + new_key_path = keystore / "root2" yk_api.setup_test_yubikey(PinManager(), str(new_key_path), reset=True) diff --git a/taf/tools/yubikey/__init__.py b/taf/tools/yubikey/__init__.py index 9d6eb0c95..5ef7c8a03 100644 --- a/taf/tools/yubikey/__init__.py +++ b/taf/tools/yubikey/__init__.py @@ -107,6 +107,22 @@ def list_keys(): return list_keys +def list_slots_command(): + @click.command( + help="Show which PIV slots on the inserted YubiKey(s) are free and which are occupied." + ) + @click.option( + "--serial", + default=None, + help="Serial number of a specific YubiKey. Lists slots for every inserted YubiKey if not specified", + ) + @catch_cli_exception(handle=YubikeyError) + def list_slots(serial): + list_yk_slots(serial) + + return list_slots + + def setup_signing_key_command(): @click.command( help="""Generate a new key on the yubikey and copy it to the given PIV slot. @@ -145,22 +161,6 @@ def setup_signing_key(certs_dir, slot, force, reset, pin_manager): return setup_signing_key -def list_slots_command(): - @click.command( - help="Show which PIV slots on the inserted YubiKey(s) are free and which are occupied." - ) - @click.option( - "--serial", - default=None, - help="Serial number of a specific YubiKey. Lists slots for every inserted YubiKey if not specified", - ) - @catch_cli_exception(handle=YubikeyError) - def list_slots(serial): - list_yk_slots(serial) - - return list_slots - - def setup_test_key_command(): @click.command( help="""Copies the specified key onto the given PIV slot of the inserted YubiKey. diff --git a/taf/tools/yubikey/yubikey_utils.py b/taf/tools/yubikey/yubikey_utils.py index 47bf5c7c5..2a061873c 100644 --- a/taf/tools/yubikey/yubikey_utils.py +++ b/taf/tools/yubikey/yubikey_utils.py @@ -6,6 +6,8 @@ from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization from taf.tuf.keys import load_signer_from_file +from yubikit.core.smartcard import ApduError, SW +from yubikit.piv import SLOT, InvalidPinError VALID_PIN = "123456" WRONG_PIN = "111111" @@ -78,8 +80,6 @@ class FakePivController: """ def __init__(self, driver): - from yubikit.piv import SLOT - self._driver = driver # slot state lives on the driver (FakeYubiKey), not here, so it # survives across separate `with _yk_piv_ctrl(...)` blocks @@ -145,11 +145,17 @@ def put_certificate(self, slot, cert): def get_certificate(self, slot): slot_data = self._slots.get(slot) if not slot_data or slot_data.get("cert") is None: - from yubikit.core.smartcard import ApduError, SW - raise ApduError(b"", SW.FILE_NOT_FOUND) return slot_data["cert"] + def get_slot_metadata(self, slot): + slot_data = self._slots.get(slot) + if not slot_data or slot_data.get("priv_key") is None: + raise ApduError(b"", SW.FILE_NOT_FOUND) + # the actual metadata fields aren't modeled - callers only care + # whether this raises or not + return object() + def reset(self): self._driver.slots.clear() @@ -170,8 +176,6 @@ def sign(self, slot, key_type, data, hash, padding): def verify_pin(self, pin): if self._driver.pin != pin: - from yubikit.piv import InvalidPinError - raise InvalidPinError(0) diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index 57dda5bf9..8685dbac1 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -19,6 +19,7 @@ from taf.yubikey.yubikey_manager import PinManager from taf.models.types import KeysMapping from ykman.device import list_all_devices +from yubikit.core import NotSupportedError from yubikit.core.smartcard import SmartCardConnection from ykman.piv import ( KEY_TYPE, @@ -654,11 +655,23 @@ def sign_piv_rsa_pkcs1v15(data, pin, serial=None): def _slot_occupied(ctrl, slot) -> bool: - """Check whether a PIV slot already has a certificate in it. - - Empty slots raise (rather than return None) when read, so absence of - an exception is what "occupied" means here. + """Check whether a PIV slot already has a key in it. + + A slot's key and certificate are separate objects, so checking only for + a certificate would miss a key left behind without one (e.g. an + interrupted setup(), or a slot provisioned by another tool). Prefer + get_slot_metadata, which reflects the key directly, and fall back to + checking for a certificate on firmware that doesn't support it (requires + 5.3+; older devices like the YubiKey 4 can't detect a key without a + certificate this way). """ + try: + ctrl.get_slot_metadata(slot) + return True + except NotSupportedError: + pass + except Exception: + return False try: ctrl.get_certificate(slot) return True @@ -666,6 +679,17 @@ def _slot_occupied(ctrl, slot) -> bool: return False +def is_slot_occupied(serial, slot) -> bool: + """Check whether a PIV slot on the inserted YubiKey already has a key in + it. See _slot_occupied for how occupancy is determined. + + Raises: + - YubikeyError + """ + with _yk_piv_ctrl(serial=serial) as [(ctrl, _)]: + return _slot_occupied(ctrl, slot) + + def get_slot_status(serial=None) -> dict: """Return the certificate (or None if empty) currently in every usable PIV slot of the inserted YubiKey(s). @@ -770,9 +794,7 @@ def setup( if mgm_key is not None: ctrl.set_management_key(MANAGEMENT_KEY_TYPE.TDES, mgm_key) else: - # Leave the management key at the PIV default rather than - # randomizing it - a forgotten random key would permanently - # block adding a key to any other slot without a full reset. + # left at the PIV default - a randomized key would block future non-reset setups mgm_key = DEFAULT_MANAGEMENT_KEY else: if management_key is None: From 04889b8f4c93ac9bb82ecf5fdad001d0d2929260 Mon Sep 17 00:00:00 2001 From: Renata Date: Tue, 14 Jul 2026 02:16:47 +0200 Subject: [PATCH 5/8] feat: store the PIV management key PIN-protected instead of at the default Reset-based setup now generates a random management key and stores it in the card's PIN-protected data object --- taf/api/yubikey.py | 50 ++++++++---- taf/tests/tuf/test_keys/test_yubikey_slots.py | 63 ++++++++++++++- taf/tools/yubikey/yubikey_utils.py | 51 ++++++++++-- taf/yubikey/yubikey.py | 78 ++++++++++++++----- 4 files changed, 197 insertions(+), 45 deletions(-) diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index 49be04d62..0429c2785 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -13,6 +13,7 @@ from taf.log import taf_logger from taf.tuf.keys import get_sslib_key_from_value from taf.tuf.repository import MAIN_ROLES +from taf.utils import get_pin_for import taf.yubikey.yubikey as yk from taf.yubikey.yubikey_manager import PinManager from yubikit.piv import SLOT @@ -38,21 +39,36 @@ def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> bool: return True +def _ensure_pin(pin_manager: PinManager, serial: str) -> str: + """Get the cached PIN for this device, prompting once for its existing + PIN (no confirmation needed, since it isn't being changed) if not + already known. The PIN is required to unlock the card's stored, + PIN-protected management key.""" + pin = pin_manager.get_pin(serial) + if pin is None: + name = f"YubiKey {serial}" if len(yk.get_serial_nums()) > 1 else "YubiKey" + pin = get_pin_for(name, confirm=False, repeat=False) + pin_manager.add_pin(serial, pin) + return pin + + def _prepare_non_reset_setup( + pin_manager: PinManager, piv_slot: SLOT, force: bool, serial: Optional[str] = None, insert_prompt: Optional[str] = None, -) -> Tuple[bool, str]: - """Resolve which YubiKey to use (unless already given) and confirm - before touching an occupied slot. Returns (proceed, serial); once - proceed is True, the target slot's occupancy has already been dealt - with, so it's safe to setup with force=True regardless of the force - that was passed in.""" +) -> Tuple[bool, str, str]: + """Resolve which YubiKey to use (unless already given), get its PIN, and + confirm before touching an occupied slot. Returns (proceed, serial, + pin); once proceed is True, the target slot's occupancy has already + been dealt with, so it's safe to setup with force=True regardless of + the force that was passed in.""" if serial is None: serial = _resolve_single_serial(insert_prompt) + pin = _ensure_pin(pin_manager, serial) proceed = _confirm_slot_overwrite(serial, piv_slot, force) - return proceed, serial + return proceed, serial, pin def _resolve_single_serial(prompt: Optional[str] = None) -> str: @@ -229,8 +245,10 @@ def get_yk_roles(path: str, serial: Optional[str] = None) -> Dict: ) def list_yk_slots(serial: Optional[str] = None) -> None: """ - Print the free/occupied status of every usable PIV slot on the inserted - YubiKey(s), including the holder name and expiry of any certificate found. + Print the free/occupied status of every PIV slot taf can set a key up + in on the inserted YubiKey(s), including the holder name and expiry of + any certificate found. Retired slots are excluded, since taf doesn't + offer them as a setup target (see SETUP_SLOTS). Arguments: serial (optional): Serial number of a specific YubiKey. Lists slots @@ -251,6 +269,8 @@ def list_yk_slots(serial: Optional[str] = None) -> None: slot_status = yk.get_slot_status(serial=dev_serial)[dev_serial] print(f"\nSerial: {dev_serial}") for slot, cert in slot_status.items(): + if slot not in SETUP_SLOTS: + continue if cert is None: print(f" {slot.name:<15} free") else: @@ -319,8 +339,10 @@ def setup_signing_yubikey( raise YubikeyError("Could not generate a new key") _, serial_num, _ = yubikeys[0] else: - # not resetting, so the PIN is untouched and not needed here - proceed, serial_num = _prepare_non_reset_setup( + # not resetting, so the PIN itself is untouched - it's only needed + # here to unlock the card's stored management key + proceed, serial_num, _ = _prepare_non_reset_setup( + pin_manager, piv_slot, force, insert_prompt="Insert the YubiKey you want to set up and press ENTER", @@ -382,7 +404,9 @@ def setup_test_yubikey( if not click.confirm("WARNING - this will reset the inserted key. Proceed?"): return else: - proceed, serial = _prepare_non_reset_setup(piv_slot, force, serial=serial) + proceed, serial, pin = _prepare_non_reset_setup( + pin_manager, piv_slot, force, serial=serial + ) if not proceed: return force = True @@ -395,8 +419,6 @@ def setup_test_yubikey( # resetting wipes the PIN too, so it's reset to the default here pin = yk.DEFAULT_PIN pin_manager.add_pin(serial, pin) - else: - pin = None pub_key = yk.setup( pin, diff --git a/taf/tests/tuf/test_keys/test_yubikey_slots.py b/taf/tests/tuf/test_keys/test_yubikey_slots.py index fb1ddf333..df650f439 100644 --- a/taf/tests/tuf/test_keys/test_yubikey_slots.py +++ b/taf/tests/tuf/test_keys/test_yubikey_slots.py @@ -1,6 +1,6 @@ import pytest from yubikit.core import NotSupportedError -from yubikit.piv import SLOT +from yubikit.piv import DEFAULT_MANAGEMENT_KEY, SLOT import taf.api.yubikey as yk_api import taf.yubikey.yubikey as yk @@ -9,6 +9,14 @@ from taf.yubikey.yubikey_manager import PinManager +def _pin_manager(fake_yubikey) -> PinManager: + """A PinManager pre-populated with the fake device's PIN, matching a + real caller that already knows it from an earlier interactive prompt.""" + pin_manager = PinManager() + pin_manager.add_pin(fake_yubikey.serial, fake_yubikey.pin) + return pin_manager + + def test_resolve_slot_rejects_retired_slots(): with pytest.raises(YubikeyError): yk_api._resolve_slot("RETIRED1") @@ -53,6 +61,49 @@ def _unsupported(self, slot): assert yk.is_slot_occupied(fake_yubikey.serial, SLOT.SIGNATURE) +def test_setup_reset_randomizes_and_stores_protected_management_key(fake_yubikey): + yk.setup( + pin="654321", + serial=fake_yubikey.serial, + cert_cn="Reset key", + reset=True, + ) + + # the management key must no longer be the well-known PIV default... + assert fake_yubikey.management_key != DEFAULT_MANAGEMENT_KEY + # ...and it must be recoverable with the PIN that was just set + with yk._yk_piv_ctrl(serial=fake_yubikey.serial) as [(ctrl, _)]: + ctrl.verify_pin("654321") + assert yk._get_protected_management_key(ctrl) == fake_yubikey.management_key + + +def test_setup_non_reset_recovers_stored_management_key(fake_yubikey, keystore): + # reset first, choosing a new PIN - this randomizes the management key + # and stores it protected by that PIN + yk.setup( + pin="654321", + serial=fake_yubikey.serial, + cert_cn="Reset key", + reset=True, + ) + + # a later non-reset call, using the same PIN, must recover the stored + # (randomized, no longer default) management key on its own and + # authenticate successfully with it - not the PIV default + new_key_pem = (keystore / "root2").read_bytes() + yk.setup( + pin="654321", + serial=fake_yubikey.serial, + cert_cn="Second key", + private_key_pem=new_key_pem, + slot=SLOT.AUTHENTICATION, + reset=False, + ) + + status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] + assert status[SLOT.AUTHENTICATION] is not None + + def test_get_piv_public_keys_tuf_covers_every_occupied_slot(fake_yubikey, keystore): new_key_pem = (keystore / "root2").read_bytes() yk.setup( @@ -133,7 +184,11 @@ def test_setup_signing_yubikey_can_write_signature_slot_without_resetting( # SIGNATURE is already occupied by the fixture's default key, so # force=True is required, exactly like any other non-reset overwrite. yk_api.setup_signing_yubikey( - PinManager(), key_size=2048, slot="SIGNATURE", reset=False, force=True + _pin_manager(fake_yubikey), + key_size=2048, + slot="SIGNATURE", + reset=False, + force=True, ) status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] @@ -150,7 +205,7 @@ def test_setup_test_yubikey_declining_occupied_slot_confirmation_leaves_key_unto new_key_path = keystore / "root2" - yk_api.setup_test_yubikey(PinManager(), str(new_key_path)) + yk_api.setup_test_yubikey(_pin_manager(fake_yubikey), str(new_key_path)) status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] original_cert = status[SLOT.SIGNATURE] @@ -168,7 +223,7 @@ def test_setup_test_yubikey_force_skips_occupied_slot_confirmation( # unmocked here. new_key_path = keystore / "root2" - yk_api.setup_test_yubikey(PinManager(), str(new_key_path), force=True) + yk_api.setup_test_yubikey(_pin_manager(fake_yubikey), str(new_key_path), force=True) status = yk.get_slot_status(serial=fake_yubikey.serial)[fake_yubikey.serial] new_cert = status[SLOT.SIGNATURE] diff --git a/taf/tools/yubikey/yubikey_utils.py b/taf/tools/yubikey/yubikey_utils.py index 2a061873c..b5be48c85 100644 --- a/taf/tools/yubikey/yubikey_utils.py +++ b/taf/tools/yubikey/yubikey_utils.py @@ -6,8 +6,9 @@ from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization from taf.tuf.keys import load_signer_from_file +from ykman.piv import OBJECT_ID_PIVMAN_PROTECTED_DATA from yubikit.core.smartcard import ApduError, SW -from yubikit.piv import SLOT, InvalidPinError +from yubikit.piv import DEFAULT_MANAGEMENT_KEY, SLOT, InvalidPinError VALID_PIN = "123456" WRONG_PIN = "111111" @@ -38,6 +39,11 @@ def __init__(self, priv_key_path, pub_key_path, scheme, serial=None, pin=VALID_P # every `with _yk_piv_ctrl(...)` block - this is what needs to # survive across separate calls (e.g. setup() then get_slot_status()) self.slots: dict = {} + # PIV general data objects (object ID -> raw bytes), used to model + # the PIN-protected management key storage (see get_object/put_object) + self.data_objects: dict = {} + self.pin_verified = False + self.management_key = DEFAULT_MANAGEMENT_KEY @property def driver(self): @@ -47,6 +53,10 @@ def driver(self): def pin(self): return self._pin + @pin.setter + def pin(self, value): + self._pin = value + @property def serial(self): return self._serial @@ -114,14 +124,17 @@ def _build_certificate(self, pub_key, priv_key): .sign(priv_key, hashes.SHA256(), default_backend()) ) - def authenticate(self, *args, **kwargs): - pass + def authenticate(self, key_type, management_key): + if management_key != self._driver.management_key: + raise ApduError(b"", SW.SECURITY_CONDITION_NOT_SATISFIED) - def set_management_key(self, *args, **kwargs): - pass + def set_management_key(self, key_type, management_key, touch=False): + self._driver.management_key = management_key - def change_pin(self, *args, **kwargs): - pass + def change_pin(self, old_pin, new_pin): + if old_pin != self._driver.pin: + raise InvalidPinError(0) + self._driver.pin = new_pin def change_puk(self, *args, **kwargs): pass @@ -158,6 +171,29 @@ def get_slot_metadata(self, slot): def reset(self): self._driver.slots.clear() + self._driver.data_objects.clear() + self._driver.pin_verified = False + self._driver.management_key = DEFAULT_MANAGEMENT_KEY + self._driver.pin = VALID_PIN + + def get_object(self, object_id): + if ( + object_id == OBJECT_ID_PIVMAN_PROTECTED_DATA + and not self._driver.pin_verified + ): + raise ApduError(b"", SW.SECURITY_CONDITION_NOT_SATISFIED) + data = self._driver.data_objects.get(object_id) + if data is None: + raise ApduError(b"", SW.FILE_NOT_FOUND) + return data + + def put_object(self, object_id, data): + if ( + object_id == OBJECT_ID_PIVMAN_PROTECTED_DATA + and not self._driver.pin_verified + ): + raise ApduError(b"", SW.SECURITY_CONDITION_NOT_SATISFIED) + self._driver.data_objects[object_id] = data def set_pin_retries(self, *args, **kwargs): pass @@ -177,6 +213,7 @@ def sign(self, slot, key_type, data, hash, padding): def verify_pin(self, pin): if self._driver.pin != pin: raise InvalidPinError(0) + self._driver.pin_verified = True class TargetYubiKey(FakeYubiKey): diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index 8685dbac1..de34d5c34 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -26,6 +26,9 @@ MANAGEMENT_KEY_TYPE, SLOT, PivSession, + generate_random_management_key, + get_pivman_protected_data, + pivman_set_mgm_key, ) from yubikit.piv import ( DEFAULT_MANAGEMENT_KEY, @@ -690,6 +693,33 @@ def is_slot_occupied(serial, slot) -> bool: return _slot_occupied(ctrl, slot) +def _store_protected_management_key(ctrl, mgm_key): + """Set a new management key and store it in the card's PIN-protected + data object, so it can be recovered later with the PIN alone instead of + needing to be remembered separately. Requires the PIN to already be + verified on this session. + """ + pivman_set_mgm_key( + ctrl, mgm_key, MANAGEMENT_KEY_TYPE.TDES, store_on_device=True + ) + taf_logger.debug("Stored new management key in protected data object.") + + +def _get_protected_management_key(ctrl): + """Return the management key stored in the card's PIN-protected data + object, or None if it was never stored there (e.g. a card set up by an + older version of taf). Requires the PIN to already be verified on this + session. + """ + try: + key = get_pivman_protected_data(ctrl).key + taf_logger.debug(f"Protected management key lookup found={key is not None}") + return key + except Exception as e: + taf_logger.debug(f"Protected management key lookup failed: {e}") + return None + + def get_slot_status(serial=None) -> dict: """Return the certificate (or None if empty) currently in every usable PIV slot of the inserted YubiKey(s). @@ -735,21 +765,23 @@ def setup( When resetting, the following steps are performed, in order: - reset to factory settings - - authenticate with the management key + - authenticate with the default management key + - generate a random management key (unless mgm_key is given) and store + it in the card's PIN-protected data object, so it can be recovered + with the PIN alone on a later non-reset call - generate key(RSA) or import given one - generate and import self-signed certificate(X509) - set pin retries - set pin - set puk(same as pin) - The management key is left at the PIV default unless mgm_key is - explicitly passed. - When not resetting, only the given slot is touched: authenticate with - the card's existing management key, generate/import the key and its - certificate into the chosen slot, and leave every other slot, the PIN, - and the PUK untouched. Refuses to overwrite an already-occupied slot - unless force=True. + the card's management key (the stored, PIN-protected one if available, + otherwise the PIV default - covering cards reset before this scheme + existed - unless management_key is explicitly given), generate/import + the key and its certificate into the chosen slot, and leave every other + slot and the PUK untouched. Refuses to overwrite an already-occupied + slot unless force=True. Args: - cert_cn(str): x509 common name @@ -757,15 +789,15 @@ def setup( - pin_retries(int): Number of retries for PIN. Only used when resetting. - private_key_pem(str): Private key in PEM format. If given, it will be imported to Yubikey. - - mgm_key(bytes): Management key to set instead of leaving it at the - PIV default. Only used when resetting. + - mgm_key(bytes): Management key to set instead of generating a random + one. Only used when resetting. - slot(SLOT): The PIV slot to write the key and certificate to. Defaults to SLOT.SIGNATURE. - reset(bool): Whether to factory-reset the card first. Defaults to False. - management_key(bytes): The card's current management key, used to - authenticate when reset=False. Defaults to - the PIV factory default management key. + authenticate when reset=False instead of + looking up the stored, PIN-protected one. - force(bool): When reset=False, whether to overwrite the target slot if it's already occupied. @@ -791,15 +823,21 @@ def setup( taf_logger.debug("Authenticating with the default management key...") ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, DEFAULT_MANAGEMENT_KEY) - if mgm_key is not None: - ctrl.set_management_key(MANAGEMENT_KEY_TYPE.TDES, mgm_key) - else: - # left at the PIV default - a randomized key would block future non-reset setups - mgm_key = DEFAULT_MANAGEMENT_KEY + # PIN is still the factory default at this point (changed further + # below), and is required to store the management key protected + ctrl.verify_pin(DEFAULT_PIN) + if mgm_key is None: + mgm_key = generate_random_management_key(MANAGEMENT_KEY_TYPE.TDES) + _store_protected_management_key(ctrl, mgm_key) else: - if management_key is None: - management_key = DEFAULT_MANAGEMENT_KEY - ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, management_key) + if management_key is not None: + ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, management_key) + else: + ctrl.verify_pin(pin) + stored_key = _get_protected_management_key(ctrl) + ctrl.authenticate( + MANAGEMENT_KEY_TYPE.TDES, stored_key or DEFAULT_MANAGEMENT_KEY + ) if not force and _slot_occupied(ctrl, slot): raise YubikeyError( f"Slot {slot.name} already has a key. Pass force=True to overwrite it." From 32aab44ddbffce5fe05ed07a78951bd5c26b87df Mon Sep 17 00:00:00 2001 From: Renata Date: Tue, 14 Jul 2026 20:10:27 +0200 Subject: [PATCH 6/8] refactor: naming/ordering cleanup and drop pytest-mock dependency --- taf/api/yubikey.py | 37 +++++++++----------- taf/tests/tuf/test_keys/test_yk.py | 18 +++++++--- taf/yubikey/yubikey.py | 56 ++++++++++++++---------------- 3 files changed, 56 insertions(+), 55 deletions(-) diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index 0429c2785..063471cd1 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -22,7 +22,7 @@ SETUP_SLOTS = {SLOT.SIGNATURE, SLOT.AUTHENTICATION, SLOT.KEY_MANAGEMENT, SLOT.CARD_AUTH} -def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> bool: +def _check_if_slot_overwrite_allowed(serial: str, piv_slot: SLOT, force: bool) -> bool: """Check whether the target slot is already occupied and, if so, confirm before overwriting it. Returns whether to proceed.""" occupied = yk.is_slot_occupied(serial, piv_slot) @@ -39,19 +39,6 @@ def _confirm_slot_overwrite(serial: str, piv_slot: SLOT, force: bool) -> bool: return True -def _ensure_pin(pin_manager: PinManager, serial: str) -> str: - """Get the cached PIN for this device, prompting once for its existing - PIN (no confirmation needed, since it isn't being changed) if not - already known. The PIN is required to unlock the card's stored, - PIN-protected management key.""" - pin = pin_manager.get_pin(serial) - if pin is None: - name = f"YubiKey {serial}" if len(yk.get_serial_nums()) > 1 else "YubiKey" - pin = get_pin_for(name, confirm=False, repeat=False) - pin_manager.add_pin(serial, pin) - return pin - - def _prepare_non_reset_setup( pin_manager: PinManager, piv_slot: SLOT, @@ -59,15 +46,19 @@ def _prepare_non_reset_setup( serial: Optional[str] = None, insert_prompt: Optional[str] = None, ) -> Tuple[bool, str, str]: - """Resolve which YubiKey to use (unless already given), get its PIN, and - confirm before touching an occupied slot. Returns (proceed, serial, - pin); once proceed is True, the target slot's occupancy has already - been dealt with, so it's safe to setup with force=True regardless of - the force that was passed in.""" + """Resolve which YubiKey to use, get its PIN, and confirm before + touching an occupied slot.""" if serial is None: serial = _resolve_single_serial(insert_prompt) - pin = _ensure_pin(pin_manager, serial) - proceed = _confirm_slot_overwrite(serial, piv_slot, force) + + # needed to unlock the card's stored, PIN-protected management key + pin = pin_manager.get_pin(serial) + if pin is None: + name = f"YubiKey {serial}" if len(yk.get_serial_nums()) > 1 else "YubiKey" + pin = get_pin_for(name, confirm=False, repeat=False) + pin_manager.add_pin(serial, pin) + + proceed = _check_if_slot_overwrite_allowed(serial, piv_slot, force) return proceed, serial, pin @@ -349,6 +340,8 @@ def setup_signing_yubikey( ) if not proceed: return + # proceed=True means any occupied-slot warning was already + # confirmed, so overwriting is now always allowed force = True key = yk.setup_new_yubikey( @@ -409,6 +402,8 @@ def setup_test_yubikey( ) if not proceed: return + # proceed=True means any occupied-slot warning was already + # confirmed, so overwriting is now always allowed force = True key_pem_path = Path(key_path) diff --git a/taf/tests/tuf/test_keys/test_yk.py b/taf/tests/tuf/test_keys/test_yk.py index 6a4e8e3f5..1fdbd49a6 100644 --- a/taf/tests/tuf/test_keys/test_yk.py +++ b/taf/tests/tuf/test_keys/test_yk.py @@ -37,12 +37,20 @@ def is_yubikey_manager_installed(): not is_yubikey_manager_installed(), reason="Yubikey Manager not installed)", ) -def test_fake_yk(mocker): +def test_fake_yk(monkeypatch): """Test public key export and signing with fake Yubikey.""" - mocker.patch("taf.yubikey.yubikey.export_piv_pub_key", return_value=_PUB) - mocker.patch("taf.yubikey.yubikey.sign_piv_rsa_pkcs1v15", return_value=_SIG) - mocker.patch("taf.yubikey.yubikey.verify_yk_inserted", return_value=True) - mocker.patch("taf.yubikey.yubikey.get_serial_nums", return_value=["1234"]) + monkeypatch.setattr( + "taf.yubikey.yubikey.export_piv_pub_key", lambda *args, **kwargs: _PUB + ) + monkeypatch.setattr( + "taf.yubikey.yubikey.sign_piv_rsa_pkcs1v15", lambda *args, **kwargs: _SIG + ) + monkeypatch.setattr( + "taf.yubikey.yubikey.verify_yk_inserted", lambda *args, **kwargs: True + ) + monkeypatch.setattr( + "taf.yubikey.yubikey.get_serial_nums", lambda *args, **kwargs: ["1234"] + ) from taf.tuf.keys import YkSigner diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index de34d5c34..5653e0981 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -657,6 +657,21 @@ def sign_piv_rsa_pkcs1v15(data, pin, serial=None): return sig +def _get_protected_management_key(ctrl): + """Return the management key stored in the card's PIN-protected data + object, or None if it was never stored there (e.g. a card set up by an + older version of taf). Requires the PIN to already be verified on this + session. + """ + try: + key = get_pivman_protected_data(ctrl).key + taf_logger.debug(f"Protected management key lookup found={key is not None}") + return key + except Exception as e: + taf_logger.debug(f"Protected management key lookup failed: {e}") + return None + + def _slot_occupied(ctrl, slot) -> bool: """Check whether a PIV slot already has a key in it. @@ -682,44 +697,16 @@ def _slot_occupied(ctrl, slot) -> bool: return False -def is_slot_occupied(serial, slot) -> bool: - """Check whether a PIV slot on the inserted YubiKey already has a key in - it. See _slot_occupied for how occupancy is determined. - - Raises: - - YubikeyError - """ - with _yk_piv_ctrl(serial=serial) as [(ctrl, _)]: - return _slot_occupied(ctrl, slot) - - def _store_protected_management_key(ctrl, mgm_key): """Set a new management key and store it in the card's PIN-protected data object, so it can be recovered later with the PIN alone instead of needing to be remembered separately. Requires the PIN to already be verified on this session. """ - pivman_set_mgm_key( - ctrl, mgm_key, MANAGEMENT_KEY_TYPE.TDES, store_on_device=True - ) + pivman_set_mgm_key(ctrl, mgm_key, MANAGEMENT_KEY_TYPE.TDES, store_on_device=True) taf_logger.debug("Stored new management key in protected data object.") -def _get_protected_management_key(ctrl): - """Return the management key stored in the card's PIN-protected data - object, or None if it was never stored there (e.g. a card set up by an - older version of taf). Requires the PIN to already be verified on this - session. - """ - try: - key = get_pivman_protected_data(ctrl).key - taf_logger.debug(f"Protected management key lookup found={key is not None}") - return key - except Exception as e: - taf_logger.debug(f"Protected management key lookup failed: {e}") - return None - - def get_slot_status(serial=None) -> dict: """Return the certificate (or None if empty) currently in every usable PIV slot of the inserted YubiKey(s). @@ -746,6 +733,17 @@ def get_slot_status(serial=None) -> dict: return status +def is_slot_occupied(serial, slot) -> bool: + """Check whether a PIV slot on the inserted YubiKey already has a key in + it. See _slot_occupied for how occupancy is determined. + + Raises: + - YubikeyError + """ + with _yk_piv_ctrl(serial=serial) as [(ctrl, _)]: + return _slot_occupied(ctrl, slot) + + @raise_yubikey_err("Cannot setup Yubikey.") def setup( pin, From a2cf6232e29041966344ec750283f373c272c256 Mon Sep 17 00:00:00 2001 From: Renata Date: Wed, 15 Jul 2026 11:49:07 +0200 Subject: [PATCH 7/8] docs: clarify protected-management-key docstrings --- taf/tools/yubikey/yubikey_utils.py | 13 +++++-------- taf/yubikey/yubikey.py | 7 ++++--- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/taf/tools/yubikey/yubikey_utils.py b/taf/tools/yubikey/yubikey_utils.py index b5be48c85..7b5e09653 100644 --- a/taf/tools/yubikey/yubikey_utils.py +++ b/taf/tools/yubikey/yubikey_utils.py @@ -79,14 +79,11 @@ def remove(self): class FakePivController: - """Fake yubikit.piv.PivSession. - - Tracks key/certificate state per PIV slot, so tests can exercise - slot-targeted setup (put_key/put_certificate/get_certificate) instead of - always seeing/writing a single fixed key regardless of which slot was - requested. SLOT.SIGNATURE starts out "occupied" with the driver's own - key/cert, matching a YubiKey that's already been set up the traditional - way; every other slot starts free. + """Fake yubikit.piv.PivSession, standing in for a real PIV connection in + tests: PIN/management-key authentication, signing, and per-slot + key/certificate storage. SLOT.SIGNATURE starts out "occupied" with the + driver's own key/cert, matching a YubiKey that's already been set up + the traditional way; every other slot starts free. """ def __init__(self, driver): diff --git a/taf/yubikey/yubikey.py b/taf/yubikey/yubikey.py index 5653e0981..e5e5c5063 100644 --- a/taf/yubikey/yubikey.py +++ b/taf/yubikey/yubikey.py @@ -659,9 +659,10 @@ def sign_piv_rsa_pkcs1v15(data, pin, serial=None): def _get_protected_management_key(ctrl): """Return the management key stored in the card's PIN-protected data - object, or None if it was never stored there (e.g. a card set up by an - older version of taf). Requires the PIN to already be verified on this - session. + object - a general-purpose PIV data object, unrelated to the 4 signing + slots taf uses - or None if it was never stored there (e.g. a card set + up by an older version of taf). Requires the PIN to already be verified + on this session. """ try: key = get_pivman_protected_data(ctrl).key From 31f792fed169bf9dfc7244c38fd9d760632cfb43 Mon Sep 17 00:00:00 2001 From: Renata Date: Fri, 17 Jul 2026 18:31:22 +0200 Subject: [PATCH 8/8] chore: update changelog --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a24f7ede7..9e734bc5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning][semver]. ### Added +- Support choosing a YubiKey PIV slot when setting up signing keys ([759]) + ### Changed - Remove unused `scheme` parameters ([757]) @@ -18,6 +20,7 @@ and this project adheres to [Semantic Versioning][semver]. - Detect signing scheme from key material instead of assuming RSA ([757]) +[759]: https://github.com/openlawlibrary/taf/pull/759 [757]: https://github.com/openlawlibrary/taf/pull/757 ## [0.39.0]