From 1ed32d387243c48f58a63e7b5eacf7aef6fbe0fc Mon Sep 17 00:00:00 2001 From: michael Date: Wed, 17 Jun 2026 16:42:50 -0700 Subject: [PATCH] feat(harness): --step-up asserts Access Document round-trip (M2G.1) --- harness/src/aliro_harness/reader/cli.py | 24 ++- harness/src/aliro_harness/reader/step_up.py | 141 +++++++++++--- harness/tests/test_reader_step_up.py | 200 +++++++++++++++++++- 3 files changed, 333 insertions(+), 32 deletions(-) diff --git a/harness/src/aliro_harness/reader/cli.py b/harness/src/aliro_harness/reader/cli.py index 837f404..53ea42d 100644 --- a/harness/src/aliro_harness/reader/cli.py +++ b/harness/src/aliro_harness/reader/cli.py @@ -12,7 +12,7 @@ from pathlib import Path import click -from aliro_harness.reader.step_up import verify_step_up_m1 +from aliro_harness.reader.step_up import verify_step_up_m2 from aliro_harness.reader.transaction import TrustBundle, run_aliro_transaction @@ -42,9 +42,10 @@ from aliro_harness.reader.transaction import TrustBundle, run_aliro_transaction @click.option( "--step-up", is_flag=True, - help="After AUTH1, also exercise Step-Up Milestone 1: SELECT ACCE5502, " - "INS=0xC9 EXCHANGE, INS=0xC3 ENVELOPE. Verifies the StepUpApplet decrypts " - "with StepUpSKReader and encrypts the empty-CBOR-map ack with StepUpSKDevice.", + help="After AUTH1, also exercise Step-Up Milestone 2: SELECT ACCE5502, " + "INS=0xC9 EXCHANGE, INS=0xC3 ENVELOPE (DeviceRequest) + GET RESPONSE " + "chaining. Decrypts the DeviceResponse under StepUpSKDevice and asserts " + "documents[0].issuerSigned.issuerAuth == access_document.bin from --trust-dir.", ) def main( trust_dir: Path | None, reader_index: int, list_readers: bool, step_up: bool @@ -87,6 +88,12 @@ def main( bundle = TrustBundle.from_trust_dir(trust_dir) click.echo(f"Loaded trust artifacts from {trust_dir}") + # Lazy: only load the AD if --step-up is set — otherwise this would + # gratuitously require access_document.bin for the AUTH1-only path. + expected_ad: bytes | None = None + if step_up: + expected_ad = (trust_dir / "access_document.bin").read_bytes() + def transmit(apdu: bytes) -> tuple[bytes, int]: data, sw1, sw2 = connection.transmit(list(apdu)) return bytes(data), (sw1 << 8) | sw2 @@ -94,7 +101,10 @@ def main( result = run_aliro_transaction(transmit=transmit, bundle=bundle) step_up_verdict: tuple[bool, str] | None = None if step_up and result.ok and result.step_up_sk is not None: - step_up_verdict = verify_step_up_m1(transmit, result.step_up_sk) + assert expected_ad is not None # set above whenever step_up is true + step_up_verdict = verify_step_up_m2( + transmit, result.step_up_sk, expected_ad + ) finally: connection.disconnect() @@ -131,9 +141,9 @@ def main( if step_up_verdict is not None: ok, msg = step_up_verdict if ok: - click.echo(f"STEP-UP M1: OK \u2014 {msg}") + click.echo(f"STEP-UP M2: OK \u2014 {msg}") else: - click.echo(f"STEP-UP M1: FAIL \u2014 {msg}", err=True) + click.echo(f"STEP-UP M2: FAIL \u2014 {msg}", err=True) raise click.exceptions.Exit(1) diff --git a/harness/src/aliro_harness/reader/step_up.py b/harness/src/aliro_harness/reader/step_up.py index a7caaec..0d40b88 100644 --- a/harness/src/aliro_harness/reader/step_up.py +++ b/harness/src/aliro_harness/reader/step_up.py @@ -1,17 +1,26 @@ -"""Step-Up Milestone 1 PC/SC verification. +"""Step-Up Milestone 2 PC/SC verification. -Verifies the M1 applet path post-AUTH1: +Drives the M2 applet path post-AUTH1: - SELECT ACCE5502 (StepUpApplet) - - INS=0xC9 EXCHANGE encrypted with StepUpSKReader -> expect SW=9000, empty body - - INS=0xC3 ENVELOPE encrypted with StepUpSKReader -> expect SW=9000, - 17B body that decrypts under StepUpSKDevice to single byte 0xA0. + - INS=0xC9 EXCHANGE encrypted with StepUpSKReader -> expect SW=9000, empty body. + (M1 behavior, unchanged in this commit; M2E.x will rework it.) + - INS=0xC3 ENVELOPE encrypted with StepUpSKReader carrying a canonical CBOR + mdoc DeviceRequest -> applet returns the first chunk of an encrypted + DeviceResponse with SW=61xx. + - INS=0xC0 GET RESPONSE repeated until SW=9000 — concatenated body decrypts + under StepUpSKDevice + deviceIv(1). + - Plaintext is a canonical CBOR DeviceResponse; we walk + ``documents[0].issuerSigned.issuerAuth`` and assert it round-trips the + Access Document the reader was provisioned with. IV layout per applet (StepUpApplet.processExchange / processEnvelope): reader -> device : 0x00*8 || counter(4B BE) device -> reader : 0x00*7 || 0x01 || counter(4B BE) -Both counters init to 1; each advances by 1 after use. +Both counters init to 1; each advances by 1 after use. The ENVELOPE response +uses deviceCounter=1 (it's the first device-side message). """ +import cbor2 from cryptography.hazmat.primitives.ciphers.aead import AESGCM from aliro_harness.reader.crypto import derive_step_up_session_keys @@ -19,12 +28,34 @@ from aliro_harness.reader.transaction import Transmit STEP_UP_AID = bytes.fromhex("A000000909ACCE5502") SW_OK = 0x9000 +SW1_MORE_DATA = 0x61 +INS_GET_RESPONSE = 0xC0 CLA_ISO = 0x00 CLA_PROPRIETARY = 0x80 INS_EXCHANGE = 0xC9 INS_ENVELOPE = 0xC3 +# Canonical-CBOR DeviceRequest the M2D.1 applet parser accepts: +# {"version":"1.0", +# "docRequests":[{"itemsRequest": }]} +# Built once at import so each call ships the same byte sequence the +# applet's DeviceRequestParserTest pins as VALID. +_DEVICE_REQUEST = cbor2.dumps( + { + "version": "1.0", + "docRequests": [ + { + "itemsRequest": cbor2.dumps( + {"docType": "org.iso.18013.5.1.mDL", "nameSpaces": {}}, + canonical=True, + ) + } + ], + }, + canonical=True, +) + def _iv_reader(counter: int) -> bytes: return b"\x00" * 8 + counter.to_bytes(4, "big") @@ -34,7 +65,36 @@ def _iv_device(counter: int) -> bytes: return b"\x00" * 7 + b"\x01" + counter.to_bytes(4, "big") -def verify_step_up_m1(transmit: Transmit, step_up_sk: bytes) -> tuple[bool, str]: +def _drain_chaining(transmit: Transmit, first_body: bytes, first_sw: int) -> tuple[bytes, int]: + """Follow ISO 7816-4 SW=61xx response chaining until a terminal SW. Returns + the concatenated body and the final SW. SW2 advertises bytes remaining + (0x00 means "256 or more"; the applet caps at 0xFF). On a non-61xx SW + we return whatever's been accumulated so far so the caller can produce a + clear error message.""" + body = bytearray(first_body) + sw = first_sw + while (sw >> 8) == SW1_MORE_DATA: + le = sw & 0xFF # 0x00 -> request 256, else the advertised count + apdu = bytes([CLA_ISO, INS_GET_RESPONSE, 0x00, 0x00, le]) + chunk, sw = transmit(apdu) + body.extend(chunk) + return bytes(body), sw + + +def verify_step_up_m2( + transmit: Transmit, step_up_sk: bytes, expected_ad: bytes +) -> tuple[bool, str]: + """Drives the M2 step-up round-trip and asserts the applet hands back our + Access Document inside the DeviceResponse. + + Args: + transmit: PC/SC transport (bytes APDU -> (response, SW)). + step_up_sk: 32 B StepUpSK from AUTH1 (decrypted derived_keys_volatile). + expected_ad: Access Document bytes the reader was provisioned with; + compared against the issuerAuth field of the decrypted DeviceResponse. + + Returns ``(ok, message)``. On failure ``message`` names the failed step. + """ sk_device, sk_reader = derive_step_up_session_keys(step_up_sk) reader_counter = 1 device_counter = 1 @@ -45,30 +105,63 @@ def verify_step_up_m1(transmit: Transmit, step_up_sk: bytes) -> tuple[bool, str] if sw != SW_OK: return False, f"SELECT ACCE5502 failed: SW=0x{sw:04X}" - # M1B.1 -- EXCHANGE: any plaintext, expect 9000+empty. - pt = b"\x00" # one byte plaintext to exercise the decrypt path + # M1B.1 -- EXCHANGE: M1 behavior (any plaintext, expect 9000+empty). M2E.x + # will rework this leg; until then we keep the M1 shape so the round-trip + # exercises both crypto contexts (EXCHANGE + ENVELOPE) under fresh keys. + pt = b"\x00" ct = AESGCM(sk_reader).encrypt(_iv_reader(reader_counter), pt, None) apdu = bytes([CLA_PROPRIETARY, INS_EXCHANGE, 0x00, 0x00, len(ct)]) + ct + b"\x00" data, sw = transmit(apdu) if sw != SW_OK: - return False, f"M1B.1 EXCHANGE failed: SW=0x{sw:04X}" + return False, f"M2 EXCHANGE failed: SW=0x{sw:04X}" if len(data) != 0: - return False, f"M1B.1 EXCHANGE expected empty body, got {len(data)}B: {data.hex()}" + return False, f"M2 EXCHANGE expected empty body, got {len(data)}B: {data.hex()}" reader_counter += 1 - # M1C.1 -- ENVELOPE: any plaintext, expect 17B response decrypting to 0xA0. - ct = AESGCM(sk_reader).encrypt(_iv_reader(reader_counter), pt, None) + # M2 -- ENVELOPE: ship a valid CBOR DeviceRequest, drain GET RESPONSE + # chaining, decrypt, and assert the round-tripped AD. + ct = AESGCM(sk_reader).encrypt(_iv_reader(reader_counter), _DEVICE_REQUEST, None) apdu = bytes([CLA_ISO, INS_ENVELOPE, 0x00, 0x00, len(ct)]) + ct + b"\x00" - data, sw = transmit(apdu) + first_body, first_sw = transmit(apdu) + full_body, sw = _drain_chaining(transmit, first_body, first_sw) if sw != SW_OK: - return False, f"M1C.1 ENVELOPE failed: SW=0x{sw:04X}" - if len(data) != 17: - return False, f"M1C.1 ENVELOPE expected 17B response, got {len(data)}B" - try: - plaintext = AESGCM(sk_device).decrypt(_iv_device(device_counter), data, None) - except Exception as e: - return False, f"M1C.1 response decrypt failed (tag/key mismatch): {e}" - if plaintext != b"\xA0": - return False, f"M1C.1 response plaintext expected 0xA0, got {plaintext.hex()}" + return False, f"M2 ENVELOPE/GET RESPONSE failed: SW=0x{sw:04X} after {len(full_body)}B" - return True, "M1 step-up verified (EXCHANGE+ENVELOPE round-trip)" + try: + plaintext = AESGCM(sk_device).decrypt(_iv_device(device_counter), full_body, None) + except Exception as e: + return False, f"M2 ENVELOPE response decrypt failed (tag/key mismatch): {e}" + + try: + resp = cbor2.loads(plaintext) + except Exception as e: + return False, f"M2 ENVELOPE plaintext is not valid CBOR: {e}" + + try: + documents = resp["documents"] + issuer_signed = documents[0]["issuerSigned"] + issuer_auth = issuer_signed["issuerAuth"] + except (KeyError, TypeError, IndexError) as e: + return False, f"M2 DeviceResponse missing documents[0].issuerSigned.issuerAuth: {e}" + + # The applet splices the AD verbatim under issuerAuth, but cbor2 decoded it + # into a 4-element list. Re-encode canonically for byte-equality with the + # provisioned AD (trustgen produces canonical CBOR ADs). Fall back to + # element-wise comparison on the decoded list to avoid false negatives + # from harmless canonical-encoding drift (e.g. tag wrappers). + ad_round_trip = cbor2.dumps(issuer_auth, canonical=True) + if ad_round_trip != expected_ad: + try: + expected_decoded = cbor2.loads(expected_ad) + except Exception: + return False, ( + "M2 issuerAuth mismatch: round-tripped bytes differ from " + f"expected AD and expected AD isn't valid CBOR ({len(expected_ad)}B)" + ) + if issuer_auth != expected_decoded: + return False, ( + "M2 issuerAuth mismatch vs. provisioned Access Document " + f"(got {len(ad_round_trip)}B, expected {len(expected_ad)}B)" + ) + + return True, "M2 step-up verified (EXCHANGE + ENVELOPE Access Document round-trip)" diff --git a/harness/tests/test_reader_step_up.py b/harness/tests/test_reader_step_up.py index 108feb0..318a11b 100644 --- a/harness/tests/test_reader_step_up.py +++ b/harness/tests/test_reader_step_up.py @@ -1,9 +1,27 @@ -"""Smoke test for step-up key derivation parity with the applet.""" +"""Smoke tests for step-up reader verification. + +Covers: + - StepUpSK -> SKDevice/SKReader HKDF parity with stdlib RFC 5869. + - M2G.1: `verify_step_up_m2` drives the full M2 ENVELOPE round-trip: + ENVELOPE(CBOR DeviceRequest) -> 252B chunk + SW=61xx + GET RESPONSE -> remainder + SW=9000 + decrypt(StepUpSKDevice, deviceIv(1)) -> canonical DeviceResponse + extract documents[0].issuerSigned.issuerAuth, assert == expected_ad. + +The M2G.1 test uses a mock transmit driven by a canned ciphertext that the +test itself synthesizes via AESGCM — so it pins the reader's framing / +chaining / decrypt / CBOR-walk logic without depending on the applet build. +M2G.2 covers the real-hardware verdict. +""" import hashlib import hmac +import cbor2 +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + from aliro_harness.reader.crypto import derive_step_up_session_keys +from aliro_harness.reader.step_up import verify_step_up_m2 def _hkdf_manual(ikm: bytes, salt: bytes, info: bytes, length: int) -> bytes: @@ -37,3 +55,183 @@ def test_derive_step_up_session_keys_matches_rfc5869(): assert sk_device == _hkdf_manual(step_up_sk, b"", b"SKDevice", 32) assert sk_reader == _hkdf_manual(step_up_sk, b"", b"SKReader", 32) + + +# ------------------------------------------------------------------------- # +# M2G.1: full ENVELOPE round-trip with GET RESPONSE chaining + AD assert. # +# ------------------------------------------------------------------------- # + + +# Same canonical-CBOR shape the applet's DeviceResponseBuilder produces and +# the M2G.1 verify function walks. Used to build a synthetic +# (decrypts-cleanly) ciphertext for the mock transmit. +def _build_fixed_device_response(ad_bytes: bytes) -> bytes: + """Mirror of applet DeviceResponseBuilder.build for the M2 fixed shape. + `issuerAuth` is the raw AD bytes spliced verbatim — we decode them so + cbor2 re-encodes canonically (matches the applet which assumes the AD + was canonical at personalization).""" + return cbor2.dumps( + { + "status": 0, + "version": "1.0", + "documents": [ + { + "docType": "org.iso.18013.5.1.mDL", + "issuerSigned": { + "issuerAuth": cbor2.loads(ad_bytes), + "nameSpaces": {}, + }, + } + ], + }, + canonical=True, + ) + + +def _fake_ad() -> bytes: + """A valid 4-element COSE_Sign1 stand-in, sized so the wrapped DeviceResponse + exceeds 252 B and forces GET RESPONSE chaining (the real applet's 388 B + case). The M2G.1 verifier doesn't crypto-verify the AD, only round-trips + its bytes — so a 4-element CBOR array of the right size is enough.""" + return cbor2.dumps( + [ + b"\xa0", # protected (empty map, bstr-wrapped) + {}, # unprotected (empty map) + b"\x11" * 200, # payload — pad to push the response over CHUNK_LEN + b"\x00" * 64, # signature (raw r||s, fixed P-256 width) + ], + canonical=True, + ) + + +def _iv_reader(counter: int) -> bytes: + return b"\x00" * 8 + counter.to_bytes(4, "big") + + +def _iv_device(counter: int) -> bytes: + return b"\x00" * 7 + b"\x01" + counter.to_bytes(4, "big") + + +class _MockTransmit: + """Records APDUs and replies from a scripted SELECT/EXCHANGE/ENVELOPE/ + GET RESPONSE sequence. ENVELOPE replies with the first 252 B of a + pre-encrypted DeviceResponse and SW=61xx; GET RESPONSE drains the rest.""" + + CHUNK_LEN = 252 + + def __init__( + self, + sk_device: bytes, + sk_reader: bytes, + device_response_plaintext: bytes, + ): + self.sk_device = sk_device + self.sk_reader = sk_reader + # Pre-encrypt the DeviceResponse under SKDevice + device-IV(counter=1). + self.ct = AESGCM(sk_device).encrypt( + _iv_device(1), device_response_plaintext, None + ) + self.ct_off = 0 + self.apdus: list[bytes] = [] + self.reader_counter = 1 + # The DeviceRequest the verifier is expected to send. + self.expected_request_pt = cbor2.dumps( + { + "version": "1.0", + "docRequests": [ + { + "itemsRequest": cbor2.dumps( + { + "docType": "org.iso.18013.5.1.mDL", + "nameSpaces": {}, + }, + canonical=True, + ) + } + ], + }, + canonical=True, + ) + + def __call__(self, apdu: bytes) -> tuple[bytes, int]: + self.apdus.append(apdu) + cla, ins = apdu[0], apdu[1] + + # SELECT 5502 (CLA=0x00 INS=0xA4 P1=0x04 P2=0x00 Lc=09 ...) + if cla == 0x00 and ins == 0xA4: + return b"", 0x9000 + + # EXCHANGE — M1 behavior: decrypt one-byte plaintext, ack 9000+empty. + if cla == 0x80 and ins == 0xC9: + # Sanity-check: payload should decrypt under reader-counter=1. + lc = apdu[4] + body = apdu[5 : 5 + lc] + AESGCM(self.sk_reader).decrypt(_iv_reader(self.reader_counter), body, None) + self.reader_counter += 1 + return b"", 0x9000 + + # ENVELOPE — decrypt DeviceRequest under reader-counter=2, ship first chunk. + if cla == 0x00 and ins == 0xC3: + lc = apdu[4] + body = apdu[5 : 5 + lc] + pt = AESGCM(self.sk_reader).decrypt( + _iv_reader(self.reader_counter), body, None + ) + assert pt == self.expected_request_pt, ( + f"ENVELOPE pt mismatch:\n got={pt.hex()}\n want={self.expected_request_pt.hex()}" + ) + self.reader_counter += 1 + chunk = self.ct[: self.CHUNK_LEN] + self.ct_off = self.CHUNK_LEN + remaining = len(self.ct) - self.CHUNK_LEN + assert remaining > 0, "test fixture must produce a chained response" + sw2 = 0xFF if remaining >= 0x100 else remaining + return chunk, 0x6100 | sw2 + + # GET RESPONSE — drain whatever's left. + if cla == 0x00 and ins == 0xC0: + chunk = self.ct[self.ct_off :] + self.ct_off = len(self.ct) + return chunk, 0x9000 + + raise AssertionError(f"unexpected APDU: {apdu.hex()}") + + +def _setup_fixture() -> tuple[bytes, bytes, _MockTransmit]: + step_up_sk = bytes(range(0xC0, 0xE0)) + sk_device, sk_reader = derive_step_up_session_keys(step_up_sk) + ad = _fake_ad() + device_response = _build_fixed_device_response(ad) + mock = _MockTransmit(sk_device, sk_reader, device_response) + return step_up_sk, ad, mock + + +def test_verify_step_up_m2_round_trips_access_document(): + step_up_sk, expected_ad, mock = _setup_fixture() + + ok, msg = verify_step_up_m2(mock, step_up_sk, expected_ad) + + assert ok, f"verify failed: {msg}" + + # Quick shape audit on the captured APDUs. + sent = mock.apdus + # SELECT, EXCHANGE, ENVELOPE, GET RESPONSE -> 4 APDUs. + assert len(sent) == 4, [a.hex() for a in sent] + assert sent[0][0:2] == bytes([0x00, 0xA4]) # SELECT + assert sent[1][0:2] == bytes([0x80, 0xC9]) # EXCHANGE + assert sent[2][0:2] == bytes([0x00, 0xC3]) # ENVELOPE + assert sent[3][0:2] == bytes([0x00, 0xC0]) # GET RESPONSE + # GET RESPONSE Le must equal the SW2 the applet handed us (136 remaining). + expected_le = len(mock.ct) - _MockTransmit.CHUNK_LEN + assert sent[3][4] == expected_le + + +def test_verify_step_up_m2_detects_ad_mismatch(): + step_up_sk, _, mock = _setup_fixture() + # Pass a different AD than the one wrapped into the canned DeviceResponse. + bogus_ad = cbor2.dumps([b"\xa0", {}, b"different", b"\x00" * 64], canonical=True) + + ok, msg = verify_step_up_m2(mock, step_up_sk, bogus_ad) + + assert not ok + assert "issuerAuth" in msg or "Access Document" in msg or "AD" in msg, msg