feat: aliro-bench-profile CLI + per-APDU latency probes
New `aliro-bench-profile` entry point drives the applet's INS_DIAG_* commands over PC/SC at two iteration counts (N=4 / N=16 by default) and computes per-op cost as the slope, factoring out APDU round-trip + any one-time Signature.init setup. Reports HMAC, ECDH, ECDSA sign, and AES-GCM 137B costs side by side, then reconstructs an AUTH1 budget using the per-primitive counts from the applet's processAuth1 flow so we can compare the reconstructed estimate against bench-test wall-clock and see how much of AUTH1 is crypto vs TLV parsing / I/O. Also extends `aliro-bench-test` to time SELECT / AUTH0 / AUTH1 / total via time.perf_counter() bracketing each transmit() call. The TransactionResult now carries a latencies_ms dict, threaded through all return sites so both success and failure paths print the new "APDU latencies (ms):" block. Made the 1.78x AUTH1 wall-clock speedup from the 4-bit GHASH commit quantifiable on real hardware (5,795 -> 3,244 ms). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -22,6 +22,7 @@ dev = [
|
||||
aliro-trustgen = "aliro_harness.trustgen.cli:main"
|
||||
aliro-personalize = "aliro_harness.personalizer.cli:main"
|
||||
aliro-bench-test = "aliro_harness.reader.cli:main"
|
||||
aliro-bench-profile = "aliro_harness.profile.cli:main"
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools>=68"]
|
||||
|
||||
8
harness/src/aliro_harness/profile/__init__.py
Normal file
8
harness/src/aliro_harness/profile/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""Drives the AliroApplet's INS_DIAG_* profiling commands.
|
||||
|
||||
Each INS_DIAG_* operation runs a single primitive (HMAC-SHA-256, ECDH,
|
||||
ECDSA-SHA-256 sign, or full AES-256-GCM encrypt of 137B) N times against
|
||||
hardcoded test vectors on the card. We measure wall-clock latency for two
|
||||
different N values; the per-op cost is the slope of the line, isolating
|
||||
APDU framing overhead from the primitive's actual cost.
|
||||
"""
|
||||
208
harness/src/aliro_harness/profile/cli.py
Normal file
208
harness/src/aliro_harness/profile/cli.py
Normal file
@@ -0,0 +1,208 @@
|
||||
"""``aliro-bench-profile`` -- per-primitive latency profiling of the Aliro
|
||||
applet on real hardware via PC/SC.
|
||||
|
||||
Selects the Aliro Expedited AID, then drives the four INS_DIAG_* commands
|
||||
the applet exposes for profiling. Each command runs an operation N times
|
||||
against hardcoded card-side test vectors. We measure two N values per
|
||||
operation and compute per-op cost as the slope, which factors out the
|
||||
fixed APDU round-trip and any one-time setup (Signature.init etc.).
|
||||
|
||||
Output: a breakdown of per-op cost in milliseconds for HMAC, ECDH, ECDSA
|
||||
sign, and full AES-256-GCM encrypt of 137 bytes -- the same primitives
|
||||
AUTH1 invokes. Plus a reconstructed AUTH1 budget computed from the
|
||||
measured per-op costs, so we can compare against the bench-test wall-clock.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
# Aliro Expedited AID -- must match AliroAids.EXPEDITED in the applet.
|
||||
EXPEDITED_AID = bytes.fromhex("A000000909ACCE5501")
|
||||
|
||||
INS_DIAG_HMAC = 0xD0
|
||||
INS_DIAG_ECDH = 0xD1
|
||||
INS_DIAG_ECDSA_SIGN = 0xD2
|
||||
INS_DIAG_GCM = 0xD3
|
||||
|
||||
# AUTH1 invokes each primitive this many times. From the applet source:
|
||||
# ECDSA verify (reader sig over Table 8-12): 1 -- modeled as ECDSA sign
|
||||
# ECDH (Kdh derivation): 1
|
||||
# HMAC (HKDF Extract + Expand for 160B): 6 -- 1 Extract + 5 Expand
|
||||
# ECDSA sign (UD signature over Table 8-13): 1
|
||||
# AES-GCM encrypt of 137B Table 8-11: 1
|
||||
AUTH1_OP_COUNTS = {
|
||||
"ecdsa_verify": 1, # use ECDSA-sign cost as proxy; same algorithm
|
||||
"ecdh": 1,
|
||||
"hmac": 6,
|
||||
"ecdsa_sign": 1,
|
||||
"gcm_137": 1,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class OpResult:
|
||||
name: str
|
||||
per_op_ms: float
|
||||
n_low: int
|
||||
t_low_ms: float
|
||||
n_high: int
|
||||
t_high_ms: float
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--reader",
|
||||
"reader_index",
|
||||
type=int,
|
||||
default=0,
|
||||
show_default=True,
|
||||
help="Index into PC/SC reader list",
|
||||
)
|
||||
@click.option(
|
||||
"--list-readers",
|
||||
is_flag=True,
|
||||
help="List PC/SC readers and exit",
|
||||
)
|
||||
@click.option(
|
||||
"--n-low",
|
||||
type=int,
|
||||
default=4,
|
||||
show_default=True,
|
||||
help="Low iteration count for slope measurement",
|
||||
)
|
||||
@click.option(
|
||||
"--n-high",
|
||||
type=int,
|
||||
default=16,
|
||||
show_default=True,
|
||||
help="High iteration count for slope measurement",
|
||||
)
|
||||
def main(reader_index: int, list_readers: bool, n_low: int, n_high: int) -> None:
|
||||
"""Profile AUTH1 sub-operations on the Aliro applet via PC/SC.
|
||||
|
||||
Card must have AliroApplet installed (personalization not required for
|
||||
diagnostics -- they use hardcoded test vectors).
|
||||
"""
|
||||
from smartcard.System import readers
|
||||
|
||||
available = readers()
|
||||
if list_readers:
|
||||
for i, r in enumerate(available):
|
||||
click.echo(f"[{i}] {r}")
|
||||
return
|
||||
|
||||
if not available:
|
||||
raise click.ClickException("No PC/SC readers found. Is pcscd running?")
|
||||
if not (0 <= reader_index < len(available)):
|
||||
raise click.ClickException(f"--reader {reader_index} out of range")
|
||||
if n_low >= n_high:
|
||||
raise click.ClickException("--n-low must be smaller than --n-high")
|
||||
if not (1 <= n_low <= 255 and 1 <= n_high <= 255):
|
||||
raise click.ClickException("N values must be in [1, 255]")
|
||||
|
||||
reader = available[reader_index]
|
||||
click.echo(f"Using reader: {reader}")
|
||||
connection = reader.createConnection()
|
||||
connection.connect()
|
||||
try:
|
||||
click.echo(f"Connected. ATR: {bytes(connection.getATR()).hex()}")
|
||||
|
||||
def transmit(apdu: bytes) -> tuple[bytes, int]:
|
||||
data, sw1, sw2 = connection.transmit(list(apdu))
|
||||
return bytes(data), (sw1 << 8) | sw2
|
||||
|
||||
# SELECT Aliro Expedited AID
|
||||
select_apdu = bytes([0x00, 0xA4, 0x04, 0x00, len(EXPEDITED_AID)]) + EXPEDITED_AID
|
||||
_, sw = transmit(select_apdu)
|
||||
if sw != 0x9000:
|
||||
raise click.ClickException(f"SELECT failed: SW=0x{sw:04X}")
|
||||
click.echo(f"SELECT Aliro Expedited OK (SW=0x{sw:04X})\n")
|
||||
|
||||
ops = [
|
||||
("hmac", INS_DIAG_HMAC, "HMAC-SHA-256 (32B key, 64B msg)"),
|
||||
("ecdh", INS_DIAG_ECDH, "ECDH P-256 (native)"),
|
||||
("ecdsa_sign", INS_DIAG_ECDSA_SIGN, "ECDSA-SHA-256 sign (native)"),
|
||||
("gcm_137", INS_DIAG_GCM, "AES-256-GCM encrypt 137B (userland)"),
|
||||
]
|
||||
|
||||
results: list[OpResult] = []
|
||||
for key, ins, label in ops:
|
||||
r = _profile_one(transmit, ins, n_low, n_high, key)
|
||||
results.append(r)
|
||||
click.echo(
|
||||
f" {label:<48} "
|
||||
f"N={n_low:>3}: {r.t_low_ms:>7.1f} ms "
|
||||
f"N={n_high:>3}: {r.t_high_ms:>7.1f} ms "
|
||||
f"-> {r.per_op_ms:>6.1f} ms/op"
|
||||
)
|
||||
|
||||
click.echo()
|
||||
_print_auth1_reconstruction(results)
|
||||
finally:
|
||||
connection.disconnect()
|
||||
|
||||
|
||||
def _profile_one(
|
||||
transmit, ins: int, n_low: int, n_high: int, name: str
|
||||
) -> OpResult:
|
||||
"""Send the diagnostic INS twice with different iteration counts and
|
||||
return per-op cost computed from the slope. The slope cancels out APDU
|
||||
overhead and any one-time setup that doesn't depend on N."""
|
||||
# APDU: CLA=80 INS=ins P1=00 P2=00 Lc=01 [N]
|
||||
def measure(n: int) -> float:
|
||||
apdu = bytes([0x80, ins, 0x00, 0x00, 0x01, n & 0xFF])
|
||||
t = time.perf_counter()
|
||||
_, sw = transmit(apdu)
|
||||
dt = (time.perf_counter() - t) * 1000.0
|
||||
if sw != 0x9000:
|
||||
raise click.ClickException(
|
||||
f"Diag INS 0x{ins:02X} (N={n}) failed: SW=0x{sw:04X}"
|
||||
)
|
||||
return dt
|
||||
|
||||
t_low = measure(n_low)
|
||||
t_high = measure(n_high)
|
||||
per_op = (t_high - t_low) / (n_high - n_low)
|
||||
return OpResult(
|
||||
name=name,
|
||||
per_op_ms=per_op,
|
||||
n_low=n_low,
|
||||
t_low_ms=t_low,
|
||||
n_high=n_high,
|
||||
t_high_ms=t_high,
|
||||
)
|
||||
|
||||
|
||||
def _print_auth1_reconstruction(results: list[OpResult]) -> None:
|
||||
"""Reconstruct an AUTH1 budget from the measured per-op costs. Compares
|
||||
the sum to the bench-test wall-clock so we can see if the primitives
|
||||
fully account for the AUTH1 latency (or if there's additional overhead
|
||||
we haven't profiled, like TLV parsing, key loading, etc.)."""
|
||||
by_name = {r.name: r.per_op_ms for r in results}
|
||||
# Use ecdsa_sign cost as ECDSA verify proxy (same algorithm, similar work).
|
||||
by_name["ecdsa_verify"] = by_name["ecdsa_sign"]
|
||||
|
||||
click.echo("AUTH1 reconstructed budget (using AUTH1_OP_COUNTS):")
|
||||
total = 0.0
|
||||
for op, count in AUTH1_OP_COUNTS.items():
|
||||
per_op = by_name.get(op, 0.0)
|
||||
sub_total = per_op * count
|
||||
total += sub_total
|
||||
click.echo(
|
||||
f" {op:<14} x {count:>2} -> {sub_total:>7.1f} ms "
|
||||
f"({per_op:.1f} ms/op)"
|
||||
)
|
||||
click.echo(f" {'estimated_total':<19} {total:>7.1f} ms")
|
||||
click.echo(
|
||||
f"\nCompare against `aliro-bench-test` AUTH1 wall-clock. Gap "
|
||||
f"= TLV parsing + key load + apdu I/O on card."
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -95,6 +95,7 @@ def main(trust_dir: Path | None, reader_index: int, list_readers: bool) -> None:
|
||||
f"RESULT: FAIL \u2014 {result.error}", err=True
|
||||
)
|
||||
_echo_tlv_breakdown(result)
|
||||
_echo_latencies(result)
|
||||
click.echo(
|
||||
"Hint: the AUTH1 response decrypted, so session keys are "
|
||||
"correct. The signature verification failed — check that the "
|
||||
@@ -103,10 +104,16 @@ def main(trust_dir: Path | None, reader_index: int, list_readers: bool) -> None:
|
||||
err=True,
|
||||
)
|
||||
raise click.exceptions.Exit(1)
|
||||
raise click.ClickException(result.error or "Aliro transaction failed.")
|
||||
# Early failure (e.g. non-9000 SW on SELECT/AUTH0/AUTH1) — print
|
||||
# whatever per-APDU latencies we collected before bailing so the
|
||||
# operator can see how long the failing step actually took.
|
||||
click.echo(f"RESULT: FAIL — {result.error}", err=True)
|
||||
_echo_latencies(result)
|
||||
raise click.exceptions.Exit(1)
|
||||
|
||||
click.echo("RESULT: OK \u2014 applet round-trip on real hardware.")
|
||||
_echo_tlv_breakdown(result)
|
||||
_echo_latencies(result)
|
||||
|
||||
|
||||
def _echo_tlv_breakdown(result) -> None:
|
||||
@@ -116,3 +123,16 @@ def _echo_tlv_breakdown(result) -> None:
|
||||
click.echo(f" 0x5A credential_PubK: {cred_pub_len}B")
|
||||
click.echo(f" 0x9E UD signature: {ud_sig_len}B")
|
||||
click.echo(f" 0x5E signaling_bitmap: 0x{bitmap_hex}")
|
||||
|
||||
|
||||
def _echo_latencies(result) -> None:
|
||||
"""Per-APDU wall-clock latencies. Surfaces card-side processing time \u2014
|
||||
the AUTH1 number is what matters for the RFAL WTX-cap analysis."""
|
||||
if not result.latencies_ms:
|
||||
return
|
||||
click.echo(" APDU latencies (ms):")
|
||||
for step in ("select", "auth0", "auth1"):
|
||||
if step in result.latencies_ms:
|
||||
click.echo(f" {step:<6} {result.latencies_ms[step]:7.1f}")
|
||||
total = sum(result.latencies_ms.values())
|
||||
click.echo(f" {'total':<6} {total:7.1f}")
|
||||
|
||||
@@ -7,8 +7,9 @@ a mocked/in-process card.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
@@ -123,6 +124,10 @@ class TransactionResult:
|
||||
ud_sig: bytes | None = None # 0x9E value (64B raw r||s)
|
||||
signaling_bitmap: bytes | None = None # 0x5E value (2B)
|
||||
error: str | None = None # populated on failure
|
||||
# Wall-clock per-APDU latencies in milliseconds. Populated as each step
|
||||
# runs, so a partial result on failure still surfaces what we measured
|
||||
# before the failing step. Keys: "select", "auth0", "auth1".
|
||||
latencies_ms: dict[str, float] = field(default_factory=dict)
|
||||
|
||||
|
||||
def run_aliro_transaction(
|
||||
@@ -140,14 +145,22 @@ def run_aliro_transaction(
|
||||
|
||||
Any non-9000 SW or signature mismatch returns ok=False with ``error`` set.
|
||||
"""
|
||||
# Collect per-APDU wall-clock latencies; mutated in place and threaded
|
||||
# through every TransactionResult so partial timings surface on failure.
|
||||
latencies: dict[str, float] = {}
|
||||
|
||||
# SELECT EXPEDITED
|
||||
select_apdu = (
|
||||
bytes([0x00, 0xA4, 0x04, 0x00, len(EXPEDITED_AID)]) + EXPEDITED_AID
|
||||
)
|
||||
_t = time.perf_counter()
|
||||
_, sw = transmit(select_apdu)
|
||||
latencies["select"] = (time.perf_counter() - _t) * 1000.0
|
||||
if sw != SW_OK:
|
||||
return TransactionResult(
|
||||
ok=False, error=f"SELECT EXPEDITED failed: SW=0x{sw:04X}"
|
||||
ok=False,
|
||||
error=f"SELECT EXPEDITED failed: SW=0x{sw:04X}",
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
# Generate reader ephemeral + transaction state
|
||||
@@ -168,9 +181,15 @@ def run_aliro_transaction(
|
||||
command_parameters=0x00,
|
||||
authentication_policy=0x00,
|
||||
)
|
||||
_t = time.perf_counter()
|
||||
auth0_resp_data, sw = transmit(build_auth0_apdu(auth0_data))
|
||||
latencies["auth0"] = (time.perf_counter() - _t) * 1000.0
|
||||
if sw != SW_OK:
|
||||
return TransactionResult(ok=False, error=f"AUTH0 failed: SW=0x{sw:04X}")
|
||||
return TransactionResult(
|
||||
ok=False,
|
||||
error=f"AUTH0 failed: SW=0x{sw:04X}",
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
cred_ephem_pub_uncompressed = find_top_level(auth0_resp_data, 0x86)
|
||||
if (
|
||||
@@ -180,6 +199,7 @@ def run_aliro_transaction(
|
||||
return TransactionResult(
|
||||
ok=False,
|
||||
error="AUTH0 response missing or malformed 0x86 credential_ePubK",
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
cred_ephem_pub_x = cred_ephem_pub_uncompressed[1:33]
|
||||
|
||||
@@ -210,16 +230,24 @@ def run_aliro_transaction(
|
||||
)
|
||||
raw_sig = sign_table_812_raw(table_812, bundle.reader_priv)
|
||||
auth1_data = build_auth1_data(raw_sig, command_parameters=0x01)
|
||||
_t = time.perf_counter()
|
||||
auth1_resp_data, sw = transmit(build_auth1_apdu(auth1_data))
|
||||
latencies["auth1"] = (time.perf_counter() - _t) * 1000.0
|
||||
if sw != SW_OK:
|
||||
return TransactionResult(ok=False, error=f"AUTH1 failed: SW=0x{sw:04X}")
|
||||
return TransactionResult(
|
||||
ok=False,
|
||||
error=f"AUTH1 failed: SW=0x{sw:04X}",
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
# Decrypt + verify
|
||||
try:
|
||||
plaintext = decrypt_auth1_response(sk_device, auth1_resp_data)
|
||||
except Exception as e:
|
||||
return TransactionResult(
|
||||
ok=False, error=f"AUTH1 response decrypt failed: {e}"
|
||||
ok=False,
|
||||
error=f"AUTH1 response decrypt failed: {e}",
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
cred_pub_5a = find_top_level(plaintext, 0x5A)
|
||||
@@ -231,12 +259,14 @@ def run_aliro_transaction(
|
||||
ok=False,
|
||||
error="Table 8-11 missing 0x9E UD signature",
|
||||
plaintext=plaintext,
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
if bitmap is None:
|
||||
return TransactionResult(
|
||||
ok=False,
|
||||
error="Table 8-11 missing 0x5E signaling_bitmap",
|
||||
plaintext=plaintext,
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
table_813 = build_table_813(
|
||||
@@ -258,6 +288,7 @@ def run_aliro_transaction(
|
||||
credential_pub=cred_pub_5a,
|
||||
ud_sig=ud_sig,
|
||||
signaling_bitmap=bitmap,
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
return TransactionResult(
|
||||
@@ -266,4 +297,5 @@ def run_aliro_transaction(
|
||||
credential_pub=cred_pub_5a,
|
||||
ud_sig=ud_sig,
|
||||
signaling_bitmap=bitmap,
|
||||
latencies_ms=latencies,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user