diff --git a/CHANGELOG.md b/CHANGELOG.md index a24f7ede..9e734bc5 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] diff --git a/taf/api/yubikey.py b/taf/api/yubikey.py index fb48e1bd..063471cd 100644 --- a/taf/api/yubikey.py +++ b/taf/api/yubikey.py @@ -1,8 +1,9 @@ from logging import DEBUG, ERROR -from typing import Dict, Optional +from typing import Dict, Optional, Tuple 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 @@ -12,8 +13,79 @@ 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 + +# 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 _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) + 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 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 True + + +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, str]: + """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) + + # 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 + + +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) @@ -121,7 +193,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,16 +213,66 @@ 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 +@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 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 + 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 slot not in SETUP_SLOTS: + continue + 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( @@ -160,14 +283,26 @@ 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: bool = False, ) -> 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 ("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 + overwrite in that case. + reset (optional): Whether to factory-reset the card first. Defaults to False. Side Effects: None @@ -175,24 +310,49 @@ def setup_signing_yubikey( Returns: None """ - if not click.confirm( - "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 yubikeys: + piv_slot = _resolve_slot(slot) + + if reset: + if not click.confirm( + "WARNING - this will delete everything from the inserted key. Proceed?" + ): + return + # resetting wipes the PIN too, so a new one is chosen here + 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 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", + ) + 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( + pin_manager, + serial_num, + key_size=key_size, + slot=piv_slot, + force=force, + reset=reset, + ) + yk.export_yk_certificate(certs_dir, key, serial_num) @log_on_start(DEBUG, "Setting up a new test YubiKey", logger=taf_logger) @@ -208,13 +368,20 @@ def setup_test_yubikey( key_path: str, key_size: Optional[int] = 2048, serial: Optional[str] = None, + slot: str = "SIGNATURE", + force: bool = False, + reset: bool = False, ) -> 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 ("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. Side Effects: None @@ -222,27 +389,43 @@ def setup_test_yubikey( Returns: None """ - 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") + piv_slot = _resolve_slot(slot) - if not click.confirm("WARNING - this will reset the inserted key. Proceed?"): - 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: + proceed, serial, pin = _prepare_non_reset_setup( + pin_manager, piv_slot, force, serial=serial + ) + if not proceed: + return + # proceed=True means any occupied-slot warning was already + # confirmed, so overwriting is now always allowed + force = True - 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...") - pin = yk.DEFAULT_PIN - pin_manager.add_pin(serial, pin) + if reset: + # resetting wipes the PIN too, so it's reset to the default here + pin = yk.DEFAULT_PIN + pin_manager.add_pin(serial, pin) 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=reset, ) print("\nPrivate key successfully imported.\n") print("\nPublic key (PEM): \n{}".format(pub_key.decode("utf-8"))) - print("Pin: {}\n".format(pin)) + if reset: + print("Pin: {}\n".format(pin)) diff --git a/taf/keys.py b/taf/keys.py index 5f4e177d..6fd7cd67 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/conftest.py b/taf/tests/tuf/test_keys/conftest.py index e3c9723e..26067047 100644 --- a/taf/tests/tuf/test_keys/conftest.py +++ b/taf/tests/tuf/test_keys/conftest.py @@ -1,6 +1,8 @@ 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 @@ -10,14 +12,22 @@ @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, 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.""" + monkeypatch.setattr(yk, "_yk_piv_ctrl", _yk_piv_ctrl_mock) + key = FakeYubiKey(keystore / "root1", keystore / "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_yk.py b/taf/tests/tuf/test_keys/test_yk.py index 6a4e8e3f..1fdbd49a 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/tests/tuf/test_keys/test_yubikey_slots.py b/taf/tests/tuf/test_keys/test_yubikey_slots.py new file mode 100644 index 00000000..df650f43 --- /dev/null +++ b/taf/tests/tuf/test_keys/test_yubikey_slots.py @@ -0,0 +1,249 @@ +import pytest +from yubikit.core import NotSupportedError +from yubikit.piv import DEFAULT_MANAGEMENT_KEY, 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 + + +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") + + +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].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_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_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( + 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, keystore): + new_key_pem = (keystore / "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 + 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, keystore): + new_key_pem = (keystore / "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_can_write_signature_slot_without_resetting( + fake_yubikey, monkeypatch +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + monkeypatch.setattr(yk, "export_yk_certificate", lambda *args, **kwargs: None) + monkeypatch.setattr("builtins.input", lambda prompt="": "New signer") + + # 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( + _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] + new_cert = status[SLOT.SIGNATURE] + assert ( + new_cert.public_key().public_numbers() != fake_yubikey.pub_key.public_numbers() + ) + + +def test_setup_test_yubikey_declining_occupied_slot_confirmation_leaves_key_untouched( + fake_yubikey, monkeypatch, keystore +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: False) + + new_key_path = keystore / "root2" + + 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] + assert ( + original_cert.public_key().public_numbers() + == fake_yubikey.pub_key.public_numbers() + ) + + +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 / "root2" + + 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] + assert ( + 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, keystore +): + monkeypatch.setattr(yk_api.click, "confirm", lambda *args, **kwargs: True) + + new_key_path = keystore / "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 074e9df9..5ef7c8a0 100644 --- a/taf/tools/yubikey/__init__.py +++ b/taf/tools/yubikey/__init__.py @@ -1,8 +1,10 @@ import click from taf.api.yubikey import ( + SETUP_SLOTS, export_yk_certificate, export_yk_public_pem, get_yk_roles, + list_yk_slots, setup_signing_yubikey, setup_test_yubikey, ) @@ -12,6 +14,8 @@ from taf.tools.repo import pin_managed from taf.yubikey.yubikey import list_connected_yubikeys, list_all_devices +SETUP_SLOT_NAMES = [s.name for s in SETUP_SLOTS] + def check_pin_command(): @click.command(help="Checks if the specified pin is valid") @@ -103,32 +107,88 @@ 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 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(SETUP_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.", + ) @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 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(SETUP_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.", + ) @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 +201,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 b43bc168..7b5e0965 100644 --- a/taf/tools/yubikey/yubikey_utils.py +++ b/taf/tools/yubikey/yubikey_utils.py @@ -6,6 +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 DEFAULT_MANAGEMENT_KEY, SLOT, InvalidPinError VALID_PIN = "123456" WRONG_PIN = "111111" @@ -31,6 +34,17 @@ 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 = {} + # 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): return self @@ -39,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 @@ -61,64 +79,138 @@ def remove(self): class FakePivController: + """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): 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 authenticate(self, *args, **kwargs): - pass - - def change_pin(self, *args, **kwargs): - pass - - def change_puk(self, *args, **kwargs): - pass - - def generate_self_signed_certificate(self, *args, **kwargs): - pass - - def get_pin_tries(self): - return 1 - - def get_certificate(self, _slot): + 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(self._driver.pub_key) + .public_key(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()) + .sign(priv_key, hashes.SHA256(), default_backend()) ) - def reset(self): + 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, key_type, management_key, touch=False): + self._driver.management_key = management_key + + 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 + + def generate_self_signed_certificate(self, *args, **kwargs): pass + def get_pin_tries(self): + return 1 + + 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, + } + + 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: + 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() + 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 + 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: - from yubikit.piv import InvalidPinError - raise InvalidPinError(0) + self._driver.pin_verified = True class TargetYubiKey(FakeYubiKey): diff --git a/taf/utils.py b/taf/utils.py index 2b9d42ad..a8af8a7d 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 d2f0f811..e5e5c506 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, @@ -26,6 +27,8 @@ SLOT, PivSession, generate_random_management_key, + get_pivman_protected_data, + pivman_set_mgm_key, ) from yubikit.piv import ( DEFAULT_MANAGEMENT_KEY, @@ -380,6 +383,39 @@ 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 +657,94 @@ 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 - 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 + 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. + + 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 + except Exception: + return False + + +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_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 + + +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, @@ -629,25 +753,52 @@ 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 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) + When not resetting, only the given slot is touched: authenticate with + 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 - 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 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 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. Returns: PIV public key in PEM format (bytes) @@ -655,16 +806,41 @@ 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() - - 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) + if reset: + taf_logger.debug( + f"Resetting YubiKey to factory settings for serial={serial}" + ) + ctrl.reset() + + taf_logger.debug("Authenticating with the default management key...") + ctrl.authenticate(MANAGEMENT_KEY_TYPE.TDES, 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 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." + ) if private_key_pem is None: taf_logger.debug("Generating RSA private key on the fly...") @@ -684,11 +860,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 +884,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 +903,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)