diff --git a/CHANGELOG.md b/CHANGELOG.md index 5130b07..41e6f0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `piv status`, `info`, `report` and `factory piv preperso status`. - `piv admin keys`: the admin security domain's key table; `--check-default` authenticates once to key version 1 with the GlobalPlatform default key. +- `piv admin rotate-keys`: replace key version 1 of the PIV security domain in + place; new keys from `PIV_SCP03_NEW_ENC` / `_MAC` / `_DEK`, key version 2 + never addressed, key check values and a fresh authentication verified after. +- `piv admin delete-factory-keyset`: delete key version 2 after a typed + confirmation; refuses while key version 1 is still the test key. - `RSA4096` in `perso generate-key`, `perso import-key`, CSR and self-signed-cert signing, and profile key mechanisms. - `perso generate-key --create-key-object`, the dev/eval fallback @@ -43,6 +48,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `factory piv preperso status` reported the SCP version of whatever applet was selected last, not the PIV security domain's. +- `cryptography>=43` is now required: the SCP02 code imports its `decrepit` + 3DES module, which older releases do not have, so an install on a system + with cryptography 41 failed at start-up. - `factory piv preperso load-config` warns that structural operations are permanent (existing elements cannot be changed or removed; only new ones can diff --git a/docs/piv/piv-commands.rst b/docs/piv/piv-commands.rst index 8677aec..25d46ce 100644 --- a/docs/piv/piv-commands.rst +++ b/docs/piv/piv-commands.rst @@ -76,6 +76,9 @@ Admin channel piv admin status the admin security domain: SCP version, key versions (no auth) piv admin keys its key table; --check-default tests key version 1 (one attempt) piv admin authenticate open the channel (mutual auth) + harmless self-test + piv admin rotate-keys replace key version 1 in place (new keys from PIV_SCP03_NEW_*) + piv admin delete-factory-keyset + IRREVERSIBLY delete key version 2, Cryptnox's keyset ``piv admin status`` selects the PIV security domain and reads its key information template. It never sends INITIALIZE UPDATE: on this chip an INITIALIZE UPDATE that is @@ -89,6 +92,29 @@ failed-authentication count clear; a wrong key costs one failed authentication, which the next successful ``piv admin authenticate`` clears. It never tries more than that one key, and skips the check when the table has no key version 1. +``piv admin rotate-keys`` replaces key version 1 of the PIV security domain in +place: one PUT KEY carrying ENC, MAC and DEK, encrypted under the current DEK, +with the key version byte left at 1. The current keys come from ``--default-keys`` +or ``PIV_SCP03_ENC`` / ``PIV_SCP03_MAC`` / ``PIV_SCP03_DEK``; the new ones from +``PIV_SCP03_NEW_ENC`` / ``PIV_SCP03_NEW_MAC`` / ``PIV_SCP03_NEW_DEK`` (hex, 16 +bytes each), never the command line. The command reads the key table first, +refuses to write anything but key version 1, compares the key check values the +card echoes with its own, authenticates once with the new keys, and re-reads the +table to confirm key version 2 is as it was. Key version 2 is Cryptnox's keyset +for remote reset and attestation; the command never addresses it. ``--dry-run`` +prints the plan with the new keys' check values and sends nothing. A refused PUT +KEY writes nothing, so the current keys still work. The new keys take the type the +domain speaks, DES on SCP02 and AES on SCP03: a domain is bound to one of the two +when it is created, and a keyset of the other type is accepted by the card yet can +never open a session afterwards. + +``piv admin delete-factory-keyset`` deletes key version 2 (DELETE KEY by key +version), authenticated with key version 1 from the ``PIV_SCP03_*`` variables. +It refuses while those are the publicly known test keys, and needs the typed +confirmation ``DELETE-FACTORY-KEYSET`` or ``--i-understand-this-is-irreversible``. +Afterwards Cryptnox's remote reset and attestation cannot reach the card; key +version 1 is the only way left to administer the PIV applet. + Certificates and objects ------------------------ diff --git a/pyproject.toml b/pyproject.toml index c323579..d726b62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,7 +34,7 @@ dependencies = [ "pyscard>=2.0", "click>=8.1", "rich>=13.0", - "cryptography>=41.0", + "cryptography>=43.0", # Cap below 6: cbor2 6.x is a Rust build with no Intel-macOS wheel, so a plain # `pip install` fails there without a Rust toolchain. Our CBOR use is trivial # (CTAP dumps/loads) and 5.x ships wheels everywhere. diff --git a/src/cryptnox_id_cli/applets/piv/ssd.py b/src/cryptnox_id_cli/applets/piv/ssd.py index ac26018..bb90081 100644 --- a/src/cryptnox_id_cli/applets/piv/ssd.py +++ b/src/cryptnox_id_cli/applets/piv/ssd.py @@ -1,17 +1,22 @@ -"""Keyless facts about the GlobalPlatform security domain that administers the PIV -applet: which SCP it speaks and which key versions it holds. +"""The GlobalPlatform security domain that administers the PIV applet: keyless facts +(which SCP it speaks, which key versions it holds) and the key-management commands +(PUT KEY, DELETE KEY) that a holder of its keys sends it. -Both reads are plain GET DATA on the selected domain. No INITIALIZE UPDATE is sent, -so nothing here counts against the card's failed-authentication limit. +The facts are plain GET DATA on the selected domain. No INITIALIZE UPDATE is sent for +them, so nothing there counts against the card's failed-authentication limit. """ from __future__ import annotations from dataclasses import dataclass +from cryptography.hazmat.decrepit.ciphers.algorithms import TripleDES +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptnox_id_cli.applets.piv.admin import SCP02, SCP03, scp_label from cryptnox_id_cli.transport.apdu import APDU, Response from cryptnox_id_cli.transport.pcsc import CardSession +from cryptnox_id_cli.transport.scp03 import Scp03Keys from cryptnox_id_cli.util import tlv PIV_SSD_AID = bytes.fromhex("A00000015153504101") @@ -26,6 +31,7 @@ TAG_KEY_INFO = 0xC0 # GlobalPlatform key type coding (GPCS 2.3, table 11-16). +KEY_TYPE_DES = 0x80 KEY_TYPE_AES = 0x88 _DES_TYPES = frozenset({0x80, 0x81, 0x82, 0x83, 0x84}) _KEY_TYPES = { @@ -137,6 +143,67 @@ def read_key_information(session: CardSession) -> Response: ) +def kcv_des(key: bytes) -> bytes: + """GP key check value for a 2-key 3DES key: 3DES-ECB of eight zero bytes, first 3 bytes.""" + enc = Cipher(TripleDES(key + key[:8]), modes.ECB()).encryptor() # noqa: S304 - KCV + return (enc.update(bytes(8)) + enc.finalize())[:3] + + +def kcv_aes(key: bytes) -> bytes: + """GP key check value for an AES key: AES-ECB of sixteen 0x01 bytes, first 3 bytes.""" + enc = Cipher(algorithms.AES(key), modes.ECB()).encryptor() # noqa: S305 - KCV + return (enc.update(b"\x01" * 16) + enc.finalize())[:3] + + +def kcv(scp_version: int, key: bytes) -> bytes: + return kcv_aes(key) if scp_version == SCP03 else kcv_des(key) + + +def encrypt_key_des(session_dek: bytes, key: bytes) -> bytes: + """SCP02 key data: 3DES-ECB under the session DEK.""" + enc = Cipher(TripleDES(session_dek + session_dek[:8]), modes.ECB()).encryptor() # noqa: S304 + return enc.update(key) + enc.finalize() + + +def encrypt_key_aes(dek: bytes, key: bytes) -> bytes: + """SCP03 key data: AES-CBC with a zero ICV under the static DEK.""" + enc = Cipher(algorithms.AES(dek), modes.CBC(bytes(16))).encryptor() + return enc.update(key) + enc.finalize() + + +def put_key_apdu( + scp_version: int, + dek: bytes, + new_keys: Scp03Keys, + *, + replace_version: int, + new_version: int, +) -> tuple[APDU, list[bytes]]: + """PUT KEY writing ENC, MAC and DEK (key IDs 1-3) as ``new_version``. + + ``replace_version`` is the version being replaced; 0 adds a new one. ``dek`` is the + key the new values travel under: the session DEK on SCP02, the static DEK on SCP03. + Returns the APDU and the three key check values the card echoes on success. + """ + blocks = b"" + kcvs: list[bytes] = [] + for key in (new_keys.enc, new_keys.mac, new_keys.dek): + check = kcv(scp_version, key) + if scp_version == SCP03: + block = bytes([KEY_TYPE_AES, 0x11, 0x10]) + encrypt_key_aes(dek, key) + else: + block = bytes([KEY_TYPE_DES, 0x10]) + encrypt_key_des(dek, key) + blocks += block + bytes([0x03]) + check + kcvs.append(check) + # P2: bit 8 set = several keys follow, low bits = first key ID. + return APDU(0x80, 0xD8, replace_version, 0x81, data=bytes([new_version]) + blocks), kcvs + + +def delete_key_version_apdu(version: int) -> APDU: + """DELETE every key of one key version (tag D2).""" + return APDU(0x80, 0xE4, 0x00, 0x00, data=bytes([0xD2, 0x01, version])) + + def describe_security_domain( session: CardSession, aids: tuple[bytes, ...] = (PIV_SSD_AID, ISD_AID) ) -> SecurityDomainInfo | None: diff --git a/src/cryptnox_id_cli/cli/commands/piv.py b/src/cryptnox_id_cli/cli/commands/piv.py index 4f28685..3954506 100644 --- a/src/cryptnox_id_cli/cli/commands/piv.py +++ b/src/cryptnox_id_cli/cli/commands/piv.py @@ -9,6 +9,7 @@ from __future__ import annotations import contextlib +import sys from collections.abc import Callable from dataclasses import dataclass from pathlib import Path @@ -34,10 +35,15 @@ from cryptnox_id_cli.applets.piv.piv import PivApplet from cryptnox_id_cli.applets.piv.slots import PIV_SLOTS from cryptnox_id_cli.applets.piv.ssd import ( + CRYPTNOX_KEY_VERSION, CUSTOMER_KEY_VERSION, PIV_SSD_AID, SecurityDomainInfo, + delete_key_version_apdu, describe_security_domain, + kcv, + put_key_apdu, + select_security_domain, ) from cryptnox_id_cli.cli.context import AppContext from cryptnox_id_cli.crypto import csr as csr_mod @@ -45,8 +51,10 @@ from cryptnox_id_cli.crypto.attestation import verify_attestation_chain from cryptnox_id_cli.output.render import state_style from cryptnox_id_cli.secrets.resolver import ( + DEFAULT_GP_KEY, MGMT_KEY_ENV, resolve_mgmt_key, + resolve_new_scp03_keys, resolve_scp03_keys, resolve_secret, ) @@ -59,6 +67,8 @@ describe_sw, ) from cryptnox_id_cli.transport.pcsc import is_contactless_interface +from cryptnox_id_cli.transport.scp02 import Scp02Session +from cryptnox_id_cli.transport.scp03 import Scp03Keys from cryptnox_id_cli.util import tlv from cryptnox_id_cli.util.hexutil import to_hex @@ -978,6 +988,256 @@ def human(c: Console) -> None: app.out.result({**domain.to_dict(), "default_key_check": check}, human) +def _open_domain_channel(session, aid: bytes, keys: Scp03Keys, key_version: int) -> PivAdmin: # noqa: ANN001 + """A secure channel to the security domain itself (not through the PIV applet), + which is where PUT KEY and DELETE KEY are processed.""" + resp = select_security_domain(session, aid) + if not resp.ok: + raise CryptnoxError(f"SELECT {aid.hex().upper()} failed (SW={resp.sw_hex()}).") + adm = PivAdmin(session) + adm.open(keys, key_version=key_version) + return adm + + +def _key_encryption_key(adm: PivAdmin, keys: Scp03Keys) -> bytes: + """SCP02 wraps new key values under the session DEK, SCP03 under the static one.""" + if isinstance(adm.scp, Scp02Session): + return adm.scp.s_dek + return keys.dek + + +def _hex_list(values: list[bytes]) -> list[str]: + return [v.hex().upper() for v in values] + + +@admin.command("rotate-keys") +@click.option( + "--default-keys", + is_flag=True, + help="The CURRENT keys are the publicly known GlobalPlatform test keys (a card as shipped).", +) +@click.option("--dry-run", is_flag=True, help="Show the plan; send nothing.") +@click.pass_obj +def admin_rotate_keys(app: AppContext, default_keys: bool, dry_run: bool) -> None: + """Replace key version 1 of the PIV security domain in place. + + Current keys: --default-keys or PIV_SCP03_ENC/MAC/DEK. New keys: PIV_SCP03_NEW_ENC/ + MAC/DEK (hex, 16 bytes each), never the command line. Key version 2, Cryptnox's + keyset for remote reset and attestation, is never addressed. + """ + new_keys = resolve_new_scp03_keys(app.redactor) + if new_keys == Scp03Keys.same(DEFAULT_GP_KEY): + raise CryptnoxError("the new keys are the publicly known test keys; choose your own.") + planning = dry_run or app.dry_run + current = None if planning else resolve_scp03_keys(app.redactor, default_keys=default_keys) + if current is not None and current == new_keys: + raise CryptnoxError("the new keys equal the current ones; nothing to rotate.") + with app.open_session() as session: + before = _require_security_domain(session) + if CUSTOMER_KEY_VERSION not in before.key_versions: + raise CryptnoxError( + f"key version {CUSTOMER_KEY_VERSION} is not in the key table of " + f"{before.aid_hex}; nothing to replace." + ) + if before.scp_version is None: + raise CryptnoxError( + f"cannot tell SCP02 from SCP03 from the key table of {before.aid_hex}; refusing." + ) + label = scp_label(before.scp_version) + new_kcvs = [kcv(before.scp_version, k) for k in (new_keys.enc, new_keys.mac, new_keys.dek)] + if planning: + plan = { + "dry_run": True, + "aid": before.aid_hex, + "scp_version": label, + "replace_version": CUSTOMER_KEY_VERSION, + "new_version": CUSTOMER_KEY_VERSION, + "new_key_kcvs": _hex_list(new_kcvs), + "key_versions": before.key_versions, + } + + def plan_human(c: Console) -> None: + c.print( + f"[bold]DRY RUN[/bold] - would replace key version {CUSTOMER_KEY_VERSION} of " + f"{before.aid_hex} ({label}) in place with one PUT KEY." + ) + c.print(f" New key check values: {' '.join(_hex_list(new_kcvs))} (ENC MAC DEK)") + c.print(f" Key versions on the card: {before.key_versions}; only 1 is addressed.") + c.print("\n [dim]Nothing was sent to the card.[/dim]") + + app.out.result(plan, plan_human) + return + + if current is None: # unreachable: resolved above whenever not planning + raise RuntimeError("current keys were not resolved") + app.out.warn( + f"this replaces key version {CUSTOMER_KEY_VERSION} of {before.aid_hex}; the current " + "keys stop working the moment the card accepts the command." + ) + if not app.yes and not click.confirm("Proceed with the rotation?", default=False): + raise click.Abort() + + adm = _open_domain_channel(session, before.aid, current, CUSTOMER_KEY_VERSION) + apdu, kcvs = put_key_apdu( + before.scp_version, + _key_encryption_key(adm, current), + new_keys, + replace_version=CUSTOMER_KEY_VERSION, + new_version=CUSTOMER_KEY_VERSION, + ) + resp = adm.send(apdu, context="PUT KEY") + if not resp.ok: + # The card writes nothing when it refuses PUT KEY; the current keys still work. + raise StatusWordError(resp.sw1, resp.sw2, context="PUT KEY (rotate-keys)") + card_kcvs_match = resp.data == bytes([CUSTOMER_KEY_VERSION]) + b"".join(kcvs) + + adm.forget_channel() + verify_error: str | None = None + try: + _open_domain_channel(session, before.aid, new_keys, CUSTOMER_KEY_VERSION) + except CryptnoxError as exc: + verify_error = str(exc) + after = describe_security_domain(session, (before.aid,)) + + factory_before = before.keys_for(CRYPTNOX_KEY_VERSION) + factory_after = after.keys_for(CRYPTNOX_KEY_VERSION) if after else None + factory_unchanged = factory_after == factory_before + payload = { + "aid": before.aid_hex, + "scp_version": label, + "replaced_version": CUSTOMER_KEY_VERSION, + "new_key_kcvs": _hex_list(kcvs), + "card_kcvs_match": card_kcvs_match, + "new_keys_authenticate": verify_error is None, + "key_versions_before": before.key_versions, + "key_versions_after": after.key_versions if after else None, + "factory_keyset_unchanged": factory_unchanged, + } + + def human(c: Console) -> None: + c.print( + f"[green]Replaced key version {CUSTOMER_KEY_VERSION}[/green] of {before.aid_hex} " + f"({label})." + ) + agree = "card agrees" if card_kcvs_match else "[red]card echoed different values[/red]" + c.print(f" Key check values: {' '.join(_hex_list(kcvs))} (ENC MAC DEK); {agree}") + auth = "yes" if verify_error is None else f"[red]no[/red] ({verify_error})" + c.print(f" New keys authenticate: {auth}") + if factory_before: + word = "unchanged" if factory_unchanged else "[red]CHANGED[/red]" + c.print(f" Key version {CRYPTNOX_KEY_VERSION} (Cryptnox): {word}") + else: + c.print(f" Key version {CRYPTNOX_KEY_VERSION} (Cryptnox): not on this card") + c.print( + " From now on set PIV_SCP03_ENC / PIV_SCP03_MAC / PIV_SCP03_DEK to the new values." + ) + + app.out.result(payload, human) + if not (card_kcvs_match and verify_error is None and factory_unchanged): + raise CryptnoxError( + "PUT KEY was accepted (SW 9000) but a post-check failed; see the lines above. " + f"Key version {CUSTOMER_KEY_VERSION} now expects the new keys: confirm with " + "`piv admin authenticate` using the PIV_SCP03_NEW_* values as PIV_SCP03_*." + ) + + +@admin.command("delete-factory-keyset") +@click.option( + "--i-understand-this-is-irreversible", + "understood", + is_flag=True, + help="Required for non-interactive use.", +) +@click.option("--dry-run", is_flag=True, help="Show the plan; send nothing.") +@click.pass_obj +def admin_delete_factory_keyset(app: AppContext, understood: bool, dry_run: bool) -> None: + """IRREVERSIBLY delete key version 2, Cryptnox's keyset, from the PIV security domain. + + Cryptnox uses key version 2 for remote reset and attestation; without it those + services cannot reach this card again. Authenticates with key version 1 from + PIV_SCP03_ENC/MAC/DEK, which must no longer be the GlobalPlatform test keys. + """ + planning = dry_run or app.dry_run + keys = None if planning else resolve_scp03_keys(app.redactor) + if keys is not None and keys == Scp03Keys.same(DEFAULT_GP_KEY): + raise CryptnoxError( + f"key version {CUSTOMER_KEY_VERSION} is still the publicly known test key; rotate " + "it first (`piv admin rotate-keys`)." + ) + with app.open_session() as session: + before = _require_security_domain(session) + factory = before.keys_for(CRYPTNOX_KEY_VERSION) + if not factory: + app.out.result( + {"aid": before.aid_hex, "deleted": False, "key_versions": before.key_versions}, + lambda c: c.print( + f"Key version {CRYPTNOX_KEY_VERSION} is not on {before.aid_hex}; nothing to " + f"delete (key versions: {before.key_versions})." + ), + ) + return + if planning: + app.out.result( + { + "dry_run": True, + "aid": before.aid_hex, + "would_delete_version": CRYPTNOX_KEY_VERSION, + "keys": [k.to_dict() for k in factory], + }, + lambda c: c.print( + f"[bold]DRY RUN[/bold] - would delete key version {CRYPTNOX_KEY_VERSION} " + f"({len(factory)} keys) from {before.aid_hex}. Nothing was sent to the card." + ), + ) + return + if keys is None: # unreachable: resolved above whenever not planning + raise RuntimeError("current keys were not resolved") + if not understood: + if app.json or not sys.stdin.isatty(): + raise CryptnoxError( + "delete-factory-keyset requires --i-understand-this-is-irreversible in " + "non-interactive mode." + ) + app.out.warn( + "This is IRREVERSIBLE: Cryptnox's remote reset and attestation will no longer " + "reach this card." + ) + if click.prompt("Type DELETE-FACTORY-KEYSET to continue") != "DELETE-FACTORY-KEYSET": + raise click.Abort() + + adm = _open_domain_channel(session, before.aid, keys, CUSTOMER_KEY_VERSION) + resp = adm.send(delete_key_version_apdu(CRYPTNOX_KEY_VERSION), context="DELETE KEY") + if not resp.ok: + raise StatusWordError(resp.sw1, resp.sw2, context="DELETE KEY (delete-factory-keyset)") + adm.forget_channel() + after = describe_security_domain(session, (before.aid,)) + + gone = after is not None and CRYPTNOX_KEY_VERSION not in after.key_versions + payload = { + "aid": before.aid_hex, + "deleted": True, + "sw": resp.sw_hex(), + "key_versions_after": after.key_versions if after else None, + "verified": gone, + } + + def human(c: Console) -> None: + c.print( + f"[green]Deleted key version {CRYPTNOX_KEY_VERSION}[/green] from {before.aid_hex} " + f"(SW={resp.sw_hex()})." + ) + remaining = after.key_versions if after else "unknown" + state = "confirmed" if gone else "[red]not confirmed[/red]" + c.print(f" Key table re-read: versions {remaining}; removal {state}.") + + app.out.result(payload, human) + if not gone: + raise CryptnoxError( + f"DELETE KEY was accepted but key version {CRYPTNOX_KEY_VERSION} is still listed; " + "re-run `piv admin keys`." + ) + + @admin.command("authenticate") @click.option( "--default-keys", diff --git a/src/cryptnox_id_cli/cli/dryrun.py b/src/cryptnox_id_cli/cli/dryrun.py index 9bddaf6..a0a0fad 100644 --- a/src/cryptnox_id_cli/cli/dryrun.py +++ b/src/cryptnox_id_cli/cli/dryrun.py @@ -36,7 +36,9 @@ "apdu transcript", "factory piv preperso finalize", "factory piv preperso load-config", + "piv admin delete-factory-keyset", "piv admin keys", + "piv admin rotate-keys", "piv perso import-key", "piv perso import-p12", "piv quickstart", diff --git a/src/cryptnox_id_cli/secrets/resolver.py b/src/cryptnox_id_cli/secrets/resolver.py index e06fb78..f1d4a91 100644 --- a/src/cryptnox_id_cli/secrets/resolver.py +++ b/src/cryptnox_id_cli/secrets/resolver.py @@ -120,6 +120,29 @@ def resolve_scp03_keys( ) +#: The replacement admin keys for a rotation, hex, 16 bytes each. +NEW_SCP03_KEY_ENVS = ("PIV_SCP03_NEW_ENC", "PIV_SCP03_NEW_MAC", "PIV_SCP03_NEW_DEK") + + +def resolve_new_scp03_keys(redactor: Redactor) -> Scp03Keys: + """The keys a rotation writes: the three ``PIV_SCP03_NEW_*`` variables, never the + command line and never a default value.""" + names = " / ".join(f"${v}" for v in NEW_SCP03_KEY_ENVS) + values: list[bytes] = [] + for name in NEW_SCP03_KEY_ENVS: + raw = os.environ.get(name) + if not raw: + raise SecretInputError( + f"The new admin keys come from {names} (hex, 16 bytes each); set all three." + ) + value = _hex_from_env(raw, f"${name}") + if len(value) != 16: + raise SecretInputError(f"${name} is {len(value)} bytes; each admin key is 16 bytes.") + redactor.register(value) + values.append(value) + return Scp03Keys(*values) + + def resolve_mgmt_key( redactor: Redactor, *, diff --git a/tests/unit/test_admin_rotate_keys.py b/tests/unit/test_admin_rotate_keys.py new file mode 100644 index 0000000..e91eb8f --- /dev/null +++ b/tests/unit/test_admin_rotate_keys.py @@ -0,0 +1,300 @@ +"""`piv admin rotate-keys` and `piv admin delete-factory-keyset`: the PUT KEY / +DELETE KEY encoding, the guards, and the post-checks. Key check values are +GlobalPlatform known answers (the default test key's 3DES KCV is the familiar 8BAF47).""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest +from _cardfakes import run, wire_session +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +from cryptnox_id_cli.applets.piv import ssd +from cryptnox_id_cli.applets.piv.admin import SCP02, SCP03 +from cryptnox_id_cli.cli.commands import piv as piv_cmd +from cryptnox_id_cli.secrets.resolver import DEFAULT_GP_KEY +from cryptnox_id_cli.transport.apdu import Response +from cryptnox_id_cli.transport.scp03 import Scp03Keys + +TEST_KEY = bytes.fromhex("0102030405060708090A0B0C0D0E0F10") +SELECT_SSD = "00A4040009A0000001515350410100" +GET_E0 = "80CA00E000" +FCI_SSD = bytes.fromhex("6F118409A00000015153504101A5049F6501FF") +E0_KVN1 = bytes.fromhex("E012C00401018010C00402018010C00403018010") +E0_KVN1_KVN2 = bytes.fromhex( + "E024C00401018010C00402018010C00403018010C00401028010C00402028010C00403028010" +) +E0_KVN2_ONLY = bytes.fromhex("E012C00401028010C00402028010C00403028010") +OK = (0x90, 0x00) + + +# ------------------------------------------------------------------ encoding --- # +def test_kcv_known_answers(): + assert ssd.kcv_des(DEFAULT_GP_KEY).hex().upper() == "8BAF47" + assert ssd.kcv_des(TEST_KEY).hex().upper() == "AD17A7" + assert ssd.kcv_aes(DEFAULT_GP_KEY).hex().upper() == "504A77" + assert ssd.kcv_aes(TEST_KEY).hex().upper() == "EDCC64" + assert ssd.kcv(SCP02, TEST_KEY) == ssd.kcv_des(TEST_KEY) + assert ssd.kcv(SCP03, TEST_KEY) == ssd.kcv_aes(TEST_KEY) + + +def test_put_key_scp03_layout_matches_the_vendor_bench(): + """[KVN] + 3 x (88 11 10 03 ), P1 = version replaced, + P2 = 81 (several keys from key ID 1).""" + apdu, kcvs = ssd.put_key_apdu( + SCP03, DEFAULT_GP_KEY, Scp03Keys.same(TEST_KEY), replace_version=1, new_version=1 + ) + assert apdu.to_bytes()[:4].hex().upper() == "80D80181" + data = apdu.data + assert data[0] == 1 + assert len(data) == 1 + 3 * 23 + enc = Cipher(algorithms.AES(DEFAULT_GP_KEY), modes.CBC(bytes(16))).encryptor() + wrapped = enc.update(TEST_KEY) + enc.finalize() + block = bytes.fromhex("881110") + wrapped + bytes([3]) + bytes.fromhex("EDCC64") + assert data[1:] == block * 3 + assert kcvs == [bytes.fromhex("EDCC64")] * 3 + + +def test_put_key_scp02_layout(): + """[KVN] + 3 x (80 10 <3DES-ECB(session DEK, key)> 03 ).""" + apdu, kcvs = ssd.put_key_apdu( + SCP02, DEFAULT_GP_KEY, Scp03Keys.same(TEST_KEY), replace_version=1, new_version=1 + ) + data = apdu.data + assert data[0] == 1 + assert len(data) == 1 + 3 * 22 + block = data[1:23] + assert block[:2].hex().upper() == "8010" + assert block[18] == 3 + assert block[19:22].hex().upper() == "AD17A7" + assert block[2:18] != TEST_KEY # travels encrypted + assert kcvs == [bytes.fromhex("AD17A7")] * 3 + + +def test_put_key_add_variant_uses_p1_zero(): + apdu, _ = ssd.put_key_apdu( + SCP02, DEFAULT_GP_KEY, Scp03Keys.same(TEST_KEY), replace_version=0, new_version=2 + ) + assert apdu.p1 == 0 + assert apdu.data[0] == 2 + + +def test_delete_key_version_apdu(): + assert ssd.delete_key_version_apdu(2).to_bytes().hex().upper() == "80E4000003D20102" + + +# ------------------------------------------------------------------- doubles --- # +class _SeqConn: + """A RawConnection answering each command from a list, the last entry sticking.""" + + def __init__(self, answers: dict[str, list[tuple[bytes, int, int]]]) -> None: + self._answers = {k.upper(): list(v) for k, v in answers.items()} + self.log: list[str] = [] + + def transmit(self, apdu: list[int]) -> tuple[list[int], int, int]: + key = bytes(apdu).hex().upper() + self.log.append(key) + seq = self._answers.get(key) + if not seq: + return [], 0x6A, 0x82 + data, sw1, sw2 = seq.pop(0) if len(seq) > 1 else seq[0] + return list(data), sw1, sw2 + + def get_atr(self) -> bytes: + return b"\x3b\x00" + + def disconnect(self) -> None: + pass + + +def _conn(*e0: bytes) -> _SeqConn: + return _SeqConn({SELECT_SSD: [(FCI_SSD, *OK)], GET_E0: [(table, *OK) for table in e0]}) + + +class _FakeAdmin: + """The secure channel double: records what was opened and sent, answers PUT KEY with + the echoed key version and check values (or a refusal) and DELETE KEY with 9000.""" + + put_key_sw = 0x9000 + echo_kcvs = True + opened: list[tuple[bytes, int]] = [] + sent: list[str] = [] + + def __init__(self, session) -> None: + self.card = session + self.scp = SimpleNamespace(s_dek=bytes(16)) + self.scp_version = SCP02 + + def open(self, keys, *, key_version: int = 0, security_level: int = 0x03) -> None: + _FakeAdmin.opened.append((keys.enc, key_version)) + + def forget_channel(self) -> None: + self.scp = None + + def send(self, apdu, *, context=None) -> Response: + _FakeAdmin.sent.append(apdu.to_bytes().hex().upper()) + if apdu.ins == 0xD8: + if _FakeAdmin.put_key_sw != 0x9000: + sw = _FakeAdmin.put_key_sw + return Response(b"", sw >> 8, sw & 0xFF) + echoed = ssd.kcv_des(TEST_KEY) * 3 if _FakeAdmin.echo_kcvs else bytes(9) + return Response(bytes([apdu.data[0]]) + echoed, 0x90, 0x00) + if apdu.ins == 0xE4: + return Response(b"", 0x90, 0x00) + return Response(b"", 0x6D, 0x00) + + +@pytest.fixture +def fake_admin(monkeypatch): + _FakeAdmin.put_key_sw = 0x9000 + _FakeAdmin.echo_kcvs = True + _FakeAdmin.opened = [] + _FakeAdmin.sent = [] + monkeypatch.setattr(piv_cmd, "PivAdmin", _FakeAdmin) + return _FakeAdmin + + +def _set_keys(monkeypatch, prefix: str, key: bytes) -> None: + for part in ("ENC", "MAC", "DEK"): + monkeypatch.setenv(f"{prefix}_{part}", key.hex()) + + +# --------------------------------------------------------------- rotate-keys --- # +def test_rotate_replaces_version_1_and_verifies(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03_NEW", TEST_KEY) + conn = _conn(E0_KVN1_KVN2) + wire_session(monkeypatch, conn) + result = run(["--json", "--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["replaced_version"] == 1 + assert payload["new_key_kcvs"] == ["AD17A7"] * 3 + assert payload["card_kcvs_match"] is True + assert payload["new_keys_authenticate"] is True + assert payload["factory_keyset_unchanged"] is True + assert payload["key_versions_after"] == [1, 2] + # Opened once with the current keys, then once with the new ones, both on version 1. + assert fake_admin.opened == [(DEFAULT_GP_KEY, 1), (TEST_KEY, 1)] + assert len(fake_admin.sent) == 1 + assert fake_admin.sent[0].startswith("80D80181") + + +def test_rotate_dry_run_sends_nothing(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03_NEW", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run(["--json", "piv", "admin", "rotate-keys", "--dry-run"]) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["dry_run"] is True + assert payload["new_key_kcvs"] == ["AD17A7"] * 3 + assert payload["scp_version"] == "SCP02" + assert fake_admin.opened == [] and fake_admin.sent == [] + + +def test_rotate_refuses_the_test_key_as_new_key(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03_NEW", DEFAULT_GP_KEY) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "publicly known test keys" in result.output + assert fake_admin.sent == [] + + +def test_rotate_refuses_without_key_version_1(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03_NEW", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN2_ONLY)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "nothing to replace" in result.output + assert fake_admin.opened == [] + + +def test_rotate_requires_the_new_keys(monkeypatch, fake_admin): + monkeypatch.delenv("PIV_SCP03_NEW_ENC", raising=False) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "PIV_SCP03_NEW_ENC" in result.output + + +def test_rotate_stops_when_the_card_refuses_put_key(monkeypatch, fake_admin): + fake_admin.put_key_sw = 0x6982 + _set_keys(monkeypatch, "PIV_SCP03_NEW", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "6982" in result.output + assert len(fake_admin.opened) == 1 # no verification attempt with the new keys + + +def test_rotate_fails_loudly_when_the_card_echoes_other_kcvs(monkeypatch, fake_admin): + fake_admin.echo_kcvs = False + _set_keys(monkeypatch, "PIV_SCP03_NEW", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "post-check failed" in result.output + + +# --------------------------------------------------- delete-factory-keyset --- # +def test_delete_removes_version_2_and_verifies(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03", TEST_KEY) + conn = _conn(E0_KVN1_KVN2, E0_KVN1) + wire_session(monkeypatch, conn) + result = run( + [ + "--json", + "piv", + "admin", + "delete-factory-keyset", + "--i-understand-this-is-irreversible", + ] + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["deleted"] is True + assert payload["verified"] is True + assert payload["key_versions_after"] == [1] + assert fake_admin.opened == [(TEST_KEY, 1)] + assert fake_admin.sent == ["80E4000003D20102"] + + +def test_delete_refuses_while_the_test_key_is_in_use(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03", DEFAULT_GP_KEY) + wire_session(monkeypatch, _conn(E0_KVN1_KVN2)) + result = run(["piv", "admin", "delete-factory-keyset", "--i-understand-this-is-irreversible"]) + assert result.exit_code != 0 + assert "rotate it first" in result.output + assert fake_admin.sent == [] + + +def test_delete_is_a_no_op_without_version_2(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN1)) + result = run( + ["--json", "piv", "admin", "delete-factory-keyset", "--i-understand-this-is-irreversible"] + ) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout)["deleted"] is False + assert fake_admin.sent == [] + + +def test_delete_needs_the_flag_when_not_interactive(monkeypatch, fake_admin): + _set_keys(monkeypatch, "PIV_SCP03", TEST_KEY) + wire_session(monkeypatch, _conn(E0_KVN1_KVN2)) + result = run(["--json", "piv", "admin", "delete-factory-keyset"]) + assert result.exit_code != 0 + assert "--i-understand-this-is-irreversible" in result.output + assert fake_admin.sent == [] + + +def test_delete_dry_run_sends_nothing(monkeypatch, fake_admin): + wire_session(monkeypatch, _conn(E0_KVN1_KVN2)) + result = run(["--json", "piv", "admin", "delete-factory-keyset", "--dry-run"]) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["would_delete_version"] == 2 + assert len(payload["keys"]) == 3 + assert fake_admin.sent == [] and fake_admin.opened == [] diff --git a/tests/unit/test_admin_rotate_scp03_loopback.py b/tests/unit/test_admin_rotate_scp03_loopback.py new file mode 100644 index 0000000..d86afb0 --- /dev/null +++ b/tests/unit/test_admin_rotate_scp03_loopback.py @@ -0,0 +1,252 @@ +"""`piv admin rotate-keys` and `delete-factory-keyset` over SCP03, end to end against a +loopback security domain: the real session (C-ENC + C-MAC), the static-DEK key +encryption and the key check values all run for real. The double keeps a key table, +answers INITIALIZE UPDATE / EXTERNAL AUTHENTICATE from its current keys, unwraps +every command, and processes PUT KEY and DELETE KEY as GlobalPlatform describes +them. It models the specification, not a vendor's security domain.""" + +from __future__ import annotations + +import json + +from _cardfakes import ResponseAdapter, run, wire_session +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +from cryptnox_id_cli.applets.piv import ssd +from cryptnox_id_cli.cli.commands import piv as piv_cmd +from cryptnox_id_cli.secrets.resolver import DEFAULT_GP_KEY +from cryptnox_id_cli.transport.apdu import APDU, Response +from cryptnox_id_cli.transport.scp03 import ( + Scp03Keys, + _aes_cmac, + _aes_ecb, + _cryptogram, + derive_session_keys, +) + +NEW_KEY = bytes.fromhex("0102030405060708090A0B0C0D0E0F10") +FACTORY_KEY = bytes.fromhex("F0E1D2C3B4A5968778695A4B3C2D1E0F") +FCI_SSD = bytes.fromhex("6F118409A00000015153504101A5049F6501FF") +CARD_CHALLENGE = bytes.fromhex("C0C1C2C3C4C5C6C7") + + +def _aes_cbc_decrypt(key: bytes, iv: bytes, data: bytes) -> bytes: + dec = Cipher(algorithms.AES(key), modes.CBC(iv)).decryptor() + return dec.update(data) + dec.finalize() + + +def _sw(sw: int) -> Response: + return Response(b"", sw >> 8, sw & 0xFF) + + +class LoopbackSecurityDomain: + """An SCP03 (i=70) security domain with a key table, selectable by the PIV SSD AID.""" + + def __init__(self, keysets: dict[int, Scp03Keys]) -> None: + self.keysets = dict(keysets) + self.pending: dict | None = None + self.session: dict | None = None + self.log: list[bytes] = [] + + # -- plain commands ----------------------------------------------------- # + def transmit(self, apdu: APDU | bytes) -> Response: + raw = apdu.to_bytes() if isinstance(apdu, APDU) else bytes(apdu) + self.log.append(raw) + cla, ins, p1, p2 = raw[:4] + if ins == 0xA4: # SELECT ends any session + self.pending = self.session = None + aid = raw[5 : 5 + raw[4]] + return Response(FCI_SSD, 0x90, 0x00) if aid == ssd.PIV_SSD_AID else _sw(0x6A82) + if ins == 0x82: # EXTERNAL AUTHENTICATE carries CLA 84 but is not yet wrapped + return self._external_authenticate(raw) + if cla & 0x04: + return self._wrapped(raw) + if ins == 0xCA and p2 == 0xE0: + return Response(self._key_information(), 0x90, 0x00) + if ins == 0x50: + return self._initialize_update(p1, raw[5:13]) + return _sw(0x6D00) + + def _key_information(self) -> bytes: + body = b"".join( + bytes([0xC0, 0x04, kid, kvn, ssd.KEY_TYPE_AES, 0x10]) + for kvn in sorted(self.keysets) + for kid in (1, 2, 3) + ) + return bytes([0xE0, len(body)]) + body + + def _initialize_update(self, kvn: int, host_challenge: bytes) -> Response: + kvn = kvn or min(self.keysets) + keys = self.keysets.get(kvn) + if keys is None: + return _sw(0x6A88) + s_enc, s_mac, _ = derive_session_keys(keys, host_challenge, CARD_CHALLENGE) + card_cryptogram = _cryptogram(s_mac, 0x00, host_challenge, CARD_CHALLENGE) + self.pending = {"kvn": kvn, "host": host_challenge, "s_enc": s_enc, "s_mac": s_mac} + body = ( + bytes(10) + + bytes([kvn, 0x03, 0x70]) + + CARD_CHALLENGE + + card_cryptogram + + bytes(3) # sequence counter + ) + return Response(body, 0x90, 0x00) + + def _external_authenticate(self, raw: bytes) -> Response: + p = self.pending + self.pending = None + if p is None: + return _sw(0x6985) + host_recv, mac_recv = raw[5:13], raw[13:21] + expected = _cryptogram(p["s_mac"], 0x01, p["host"], CARD_CHALLENGE) + full = _aes_cmac(p["s_mac"], bytes(16) + raw[:5] + host_recv) + if host_recv != expected or mac_recv != full[:8]: + return _sw(0x6982) + self.session = {**p, "chain": full, "counter": 1} + return _sw(0x9000) + + # -- secured commands --------------------------------------------------- # + def _wrapped(self, raw: bytes) -> Response: + s = self.session + if s is None: + return _sw(0x6985) + header, lc = raw[:5], raw[4] + body = raw[5 : 5 + lc] + data, mac = body[:-8], body[-8:] + full = _aes_cmac(s["s_mac"], s["chain"] + header + data) + if mac != full[:8]: + return _sw(0x6988) + s["chain"] = full + if data: + icv = _aes_ecb(s["s_enc"], s["counter"].to_bytes(16, "big")) + plain = _aes_cbc_decrypt(s["s_enc"], icv, data) + data = plain[: plain.rindex(b"\x80")] + s["counter"] += 1 + ins, p1, p2 = raw[1], raw[2], raw[3] + if ins == 0xD8: + return self._put_key(p1, p2, data) + if ins == 0xE4: + return self._delete_key(data) + return _sw(0x6D00) + + def _put_key(self, p1: int, p2: int, data: bytes) -> Response: + if p2 != 0x81: + return _sw(0x6A86) + dek = self.keysets[self.session["kvn"]].dek # type: ignore[index] + new_version, off, keys = data[0], 1, [] + while off < len(data): + key_type, block_len = data[off], data[off + 1] + block = data[off + 2 : off + 2 + block_len] + off += 2 + block_len + kcv_len = data[off] + check = data[off + 1 : off + 1 + kcv_len] + off += 1 + kcv_len + if key_type != ssd.KEY_TYPE_AES or block_len != 0x11 or block[0] != 0x10: + return _sw(0x6A80) + key = _aes_cbc_decrypt(dek, bytes(16), block[1:17]) + if ssd.kcv_aes(key) != check: + return _sw(0x6982) # nothing written + keys.append(key) + if len(keys) != 3: + return _sw(0x6A80) + if p1 == 0 and new_version in self.keysets: + return _sw(0x6A80) + if p1 and p1 not in self.keysets: + return _sw(0x6A88) + self.keysets.pop(p1, None) + self.keysets[new_version] = Scp03Keys(*keys) + echoed = bytes([new_version]) + b"".join(ssd.kcv_aes(k) for k in keys) + return Response(echoed, 0x90, 0x00) + + def _delete_key(self, data: bytes) -> Response: + if data[:2] != b"\xd2\x01": + return _sw(0x6A80) + version = data[2] + if version not in self.keysets: + return _sw(0x6A88) + if version == self.session["kvn"]: # type: ignore[index] + return _sw(0x6985) + del self.keysets[version] + return _sw(0x9000) + + +def _domain() -> LoopbackSecurityDomain: + return LoopbackSecurityDomain( + {1: Scp03Keys.same(DEFAULT_GP_KEY), 2: Scp03Keys.same(FACTORY_KEY)} + ) + + +def _set_keys(monkeypatch, prefix: str, key: bytes) -> None: + for part in ("ENC", "MAC", "DEK"): + monkeypatch.setenv(f"{prefix}_{part}", key.hex()) + + +def _wired(monkeypatch, domain: LoopbackSecurityDomain) -> None: + wire_session(monkeypatch, ResponseAdapter(domain)) + + +def test_rotate_keys_over_scp03(monkeypatch): + domain = _domain() + _wired(monkeypatch, domain) + _set_keys(monkeypatch, "PIV_SCP03_NEW", NEW_KEY) + result = run(["--json", "--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["scp_version"] == "SCP03" + assert payload["new_key_kcvs"] == ["EDCC64"] * 3 + assert payload["card_kcvs_match"] is True + assert payload["new_keys_authenticate"] is True + assert payload["factory_keyset_unchanged"] is True + assert payload["key_versions_after"] == [1, 2] + assert domain.keysets[1] == Scp03Keys.same(NEW_KEY) + assert domain.keysets[2] == Scp03Keys.same(FACTORY_KEY) + put_keys = [f for f in domain.log if f[1] == 0xD8] + assert len(put_keys) == 1 + assert put_keys[0][0] == 0x84 # wrapped + assert not any(NEW_KEY in frame for frame in domain.log) # never in clear on the wire + + +def test_rotate_keys_with_the_wrong_dek_writes_nothing(monkeypatch): + domain = _domain() + _wired(monkeypatch, domain) + _set_keys(monkeypatch, "PIV_SCP03_NEW", NEW_KEY) + monkeypatch.setattr(piv_cmd, "_key_encryption_key", lambda adm, keys: bytes(16)) + result = run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]) + assert result.exit_code != 0 + assert "6982" in result.output + assert domain.keysets[1] == Scp03Keys.same(DEFAULT_GP_KEY) + + +def test_delete_factory_keyset_over_scp03(monkeypatch): + domain = LoopbackSecurityDomain({1: Scp03Keys.same(NEW_KEY), 2: Scp03Keys.same(FACTORY_KEY)}) + _wired(monkeypatch, domain) + _set_keys(monkeypatch, "PIV_SCP03", NEW_KEY) + result = run( + [ + "--json", + "piv", + "admin", + "delete-factory-keyset", + "--i-understand-this-is-irreversible", + ] + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["verified"] is True + assert payload["key_versions_after"] == [1] + assert list(domain.keysets) == [1] + deletes = [f for f in domain.log if f[1] == 0xE4] + assert len(deletes) == 1 and deletes[0][0] == 0x84 + + +def test_customer_flow_rotate_then_delete(monkeypatch): + """The documented sequence on one card: rotate with the shipped keys, then drop the + factory keyset with the new ones.""" + domain = _domain() + _wired(monkeypatch, domain) + _set_keys(monkeypatch, "PIV_SCP03_NEW", NEW_KEY) + assert run(["--yes", "piv", "admin", "rotate-keys", "--default-keys"]).exit_code == 0 + _set_keys(monkeypatch, "PIV_SCP03", NEW_KEY) + result = run(["piv", "admin", "delete-factory-keyset", "--i-understand-this-is-irreversible"]) + assert result.exit_code == 0, result.output + assert domain.keysets == {1: Scp03Keys.same(NEW_KEY)}