Share the card doubles behind one adapter

tests/unit/_cardfakes.py

@@ -0,0 +1,154 @@
+"""Shared card doubles and wiring helpers.
+
+Two transmit conventions exist in this suite and both are deliberate:
+
+* ``RawConnection`` style, ``transmit(list[int]) -> (list[int], sw1, sw2)``. This is what
+  ``CardSession`` drives, and what most doubles here implement.
+* transmit-callable style, ``transmit(APDU | bytes) -> Response``. The SCP doubles use it
+  because ``open_channel`` takes a bare callable rather than a connection.
+
+:class:`ResponseAdapter` bridges the second into the first, which is what lets a secure
+channel double sit behind ``connect()`` and be driven by a real ``CardSession`` - redactor,
+APDU log and 6Cxx/61xx chaining all running for real.
+
+Import this as ``_cardfakes`` from any module in this directory. There is no
+``__init__.py`` here, so pytest's prepend import mode puts the directory on ``sys.path``
+under both ``python -m pytest`` and the bare ``pytest`` entry point CI uses. The leading
+underscore keeps it out of collection.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Sequence
+
+from click.testing import CliRunner
+
+from cryptnox_id_cli.cli.context import AppContext
+from cryptnox_id_cli.cli.main import main as root
+
+_Reply = tuple[list[int], int, int]
+
+
+class QueueConn:
+    """A RawConnection returning queued ``(data, sw1, sw2)`` replies in order."""
+
+    def __init__(self, responses: Iterable[tuple[Sequence[int] | bytes, int, int]]) -> None:
+        self._responses = list(responses)
+        self.sent: list[bytes] = []
+
+    def transmit(self, apdu: list[int]) -> _Reply:
+        self.sent.append(bytes(apdu))
+        data, sw1, sw2 = self._responses.pop(0)
+        return list(data), sw1, sw2
+
+    def get_atr(self) -> bytes:
+        return b""
+
+    def disconnect(self) -> None:
+        pass
+
+
+class RecordingConn:
+    """A RawConnection that records every frame and answers with one fixed reply.
+
+    Useful for asserting on command shaping (chaining, P1/P2, Lc) rather than on state.
+    """
+
+    def __init__(self, reply: _Reply = ([], 0x90, 0x00), *, atr: bytes = b"\x3b\x00") -> None:
+        self.frames: list[bytes] = []
+        self._reply = reply
+        self._atr = atr
+
+    def transmit(self, apdu: list[int]) -> _Reply:
+        self.frames.append(bytes(apdu))
+        data, sw1, sw2 = self._reply
+        return list(data), sw1, sw2
+
+    def get_atr(self) -> bytes:
+        return self._atr
+
+    def disconnect(self) -> None:
+        pass
+
+
+class ResponseAdapter:
+    """Expose a transmit-callable card as a RawConnection.
+
+    ``answers`` is consulted first, keyed on the full command hex exactly like conftest's
+    ``MockConnection``. That covers framing the wrapped double does not model - a SELECT,
+    say - without teaching it new commands. Anything unmatched goes to the card, whose
+    reply may be a ``Response`` or an already-raw tuple.
+
+    ``log`` holds every command as uppercase hex, which is usually the assertion you want:
+    it catches command-shaping regressions that a state-modelling fake would hide.
+    """
+
+    def __init__(
+        self,
+        card: object,
+        *,
+        atr: bytes = b"\x3b\x00",
+        answers: dict[str, _Reply] | None = None,
+    ) -> None:
+        self._card = card
+        self._atr = atr
+        self._answers = {k.upper(): v for k, v in (answers or {}).items()}
+        self.log: list[str] = []
+
+    def transmit(self, apdu: list[int]) -> _Reply:
+        raw = bytes(apdu)
+        key = raw.hex().upper()
+        self.log.append(key)
+        scripted = self._answers.get(key)
+        if scripted is not None:
+            data, sw1, sw2 = scripted
+            return list(data), sw1, sw2
+        reply = self._card.transmit(raw)  # type: ignore[attr-defined]
+        if isinstance(reply, tuple):
+            data, sw1, sw2 = reply
+            return list(data), sw1, sw2
+        return list(reply.data), reply.sw1, reply.sw2
+
+    def get_atr(self) -> bytes:
+        return self._atr
+
+    def disconnect(self) -> None:
+        pass
+
+
+def wire_session(monkeypatch, conn, *, reader_name: str | None = None) -> None:
+    """Route every ``app.open_session()`` at ``conn``.
+
+    This is the seam 63 call sites share. ``make_session`` still runs, so the redactor,
+    the ``--apdu-log`` file and the session's chaining are exercised rather than bypassed.
+    Commands that import ``connect``/``pick_reader`` directly (``doctor``, ``readers``)
+    are not covered by this - use :func:`wire_pcsc` for those.
+    """
+    monkeypatch.setattr(
+        AppContext,
+        "open_session",
+        lambda self: self.make_session(conn, reader_name=reader_name),
+    )
+
+
+def wire_pcsc(monkeypatch, module, conn, *, reader_name: str, infos=None) -> None:
+    """Substitute the PC/SC names a command module imported into its own namespace."""
+    if infos is not None:
+        monkeypatch.setattr(module, "reader_states", lambda: infos, raising=True)
+    monkeypatch.setattr(module, "pick_reader", lambda preference: reader_name, raising=True)
+    monkeypatch.setattr(module, "connect", lambda name: conn, raising=True)
+
+
+def run(args: Sequence[str], **kwargs):
+    """Invoke the CLI through CliRunner."""
+    return CliRunner().invoke(root, list(args), **kwargs)
+
+
+def run_both_modes(args: Sequence[str], **kwargs):
+    """Return ``(json_result, human_result)`` for the same command.
+
+    About a fifth of the missed statements in the command modules sit inside the
+    ``human(c)`` closures, which only run when ``--json`` is absent, so the second
+    invocation is usually the cheapest coverage in any wave.
+    """
+    return run(["--json", *args], **kwargs), run(list(args), **kwargs)

tests/unit/test_card_fakes_seam.py

@@ -0,0 +1,117 @@
+"""The seam the card doubles hang on, and the boundary that keeps it honest.
+
+Substituting the card happens in one of two places, and which one depends on how the
+command module reached PC/SC:
+
+* ``AppContext.open_session`` for the 63 call sites that go through it, and
+* the command module's own namespace for the two that do ``from ... import connect``.
+
+A command added later that imports the same way would slip past a fixture patched only
+at ``open_session`` and reach a real reader in CI, so the boundary is asserted here
+rather than left as a convention.
+"""
+
+import ast
+import pathlib
+
+import pytest
+from _cardfakes import ResponseAdapter, run, wire_session
+from test_scp03 import GP_KEY, FakeScp03Card
+
+from cryptnox_id_cli.transport.scp03 import Scp03Keys
+
+SRC = pathlib.Path(__file__).resolve().parents[2] / "src" / "cryptnox_id_cli"
+
+#: Modules allowed to bind PC/SC entry points into their own namespace. Everything else
+#: must go through AppContext.open_session, or a fixture cannot substitute the card.
+PCSC_IMPORTERS = {
+    "cli/context.py",
+    "cli/commands/doctor.py",
+    "cli/commands/readers.py",
+}
+PCSC_NAMES = {"connect", "pick_reader", "reader_states", "PCSCConnection"}
+
+
+def _modules_importing_pcsc_names() -> set[str]:
+    found = set()
+    for path in SRC.rglob("*.py"):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for node in ast.walk(tree):
+            if (
+                isinstance(node, ast.ImportFrom)
+                and (node.module or "").endswith("transport.pcsc")
+                and {a.name for a in node.names} & PCSC_NAMES
+            ):
+                found.add(path.relative_to(SRC).as_posix())
+    return found
+
+
+def test_only_known_modules_bind_pcsc_entry_points():
+    assert _modules_importing_pcsc_names() == PCSC_IMPORTERS, (
+        "A module started importing PC/SC entry points directly. Either route it through "
+        "AppContext.open_session, or add it here and teach the fixtures to patch it - "
+        "otherwise its tests will reach a real reader."
+    )
+
+
+def test_smartcard_is_imported_only_inside_the_transport_wrapper():
+    """pyscard stays behind transport/pcsc.py, so substituting there is sufficient."""
+    offenders = set()
+    for path in SRC.rglob("*.py"):
+        if path.relative_to(SRC).as_posix() == "transport/pcsc.py":
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for node in ast.walk(tree):
+            mod = ""
+            if isinstance(node, ast.ImportFrom):
+                mod = node.module or ""
+            elif isinstance(node, ast.Import):
+                mod = node.names[0].name
+            if mod == "smartcard" or mod.startswith("smartcard."):
+                offenders.add(path.relative_to(SRC).as_posix())
+    assert offenders == set()
+
+
+@pytest.fixture
+def scp03_card(monkeypatch):
+    """A real SCP03 responder behind connect(), reached through open_session.
+
+    The point of the adapter: os.urandom is NOT patched. The host challenge is fresh on
+    every run and the card derives its cryptogram from the one it receives, which is
+    exactly what a recorded transcript cannot do.
+    """
+    card = FakeScp03Card(Scp03Keys.same(GP_KEY), bytes.fromhex("1122334455667788"))
+    conn = ResponseAdapter(
+        card,
+        # The SCP double models the secure channel, not applet selection.
+        answers={"00A404000BA00000030800001000010000": ([0x6F, 0x00], 0x90, 0x00)},
+    )
+    wire_session(monkeypatch, conn, reader_name="Fake Reader 00 00")
+    return conn
+
+
+def test_admin_authenticate_completes_a_real_mutual_authentication(scp03_card):
+    result = run(["--json", "piv", "admin", "authenticate", "--default-keys"])
+    assert result.exit_code == 0, result.output
+    assert '"authenticated": true' in result.output
+    assert "SCP03" in result.output
+
+
+def test_the_channel_really_was_negotiated_not_stubbed(scp03_card):
+    """Guards against a double that just answers 9000: the wire must show the handshake."""
+    run(["--json", "piv", "admin", "authenticate", "--default-keys"])
+    log = scp03_card.log
+    assert any(c.startswith("8050") for c in log), "no INITIALIZE UPDATE"
+    assert any(c.startswith("8482") for c in log), "no EXTERNAL AUTHENTICATE"
+    # The self-test rides the open channel: CLA 0x04 marks a C-MAC/C-ENC wrapped command.
+    assert any(c.startswith("04") for c in log), "nothing was sent over the open channel"
+
+
+def test_a_fresh_host_challenge_is_used_each_run(scp03_card, monkeypatch):
+    """The handshake is computed, not replayed, so two runs differ on the wire."""
+    run(["--json", "piv", "admin", "authenticate", "--default-keys"])
+    first = next(c for c in scp03_card.log if c.startswith("8050"))
+    scp03_card.log.clear()
+    run(["--json", "piv", "admin", "authenticate", "--default-keys"])
+    second = next(c for c in scp03_card.log if c.startswith("8050"))
+    assert first != second

tests/unit/test_desfire.py

@@ -1,6 +1,7 @@
 """DESFire transport tests: native framing, 91AF frame chaining, parsers, errors."""
 
 import pytest
+from _cardfakes import QueueConn
 
 from cryptnox_id_cli.applets.mifare import desfire
 from cryptnox_id_cli.applets.mifare.desfire import (
@@ -11,25 +12,6 @@
 from cryptnox_id_cli.transport.pcsc import CardSession
 
 
-class QueueConn:
-    """A RawConnection that returns queued (data, sw1, sw2) responses in order."""
-
-    def __init__(self, responses):
-        self._responses = list(responses)
-        self.sent: list[bytes] = []
-
-    def transmit(self, apdu):
-        self.sent.append(bytes(apdu))
-        data, sw1, sw2 = self._responses.pop(0)
-        return list(data), sw1, sw2
-
-    def get_atr(self):
-        return b""
-
-    def disconnect(self):
-        pass
-
-
 def test_frame_encoding():
     assert DesfireTransport._frame(0x60).hex().upper() == "9060000000"
     assert (

tests/unit/test_ev2.py

@@ -1,6 +1,7 @@
 """DESFire EV2 auth + secure-messaging tests (offline; the card is the final arbiter)."""
 
 import pytest
+from _cardfakes import QueueConn
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from cryptography.hazmat.primitives.cmac import CMAC
 
@@ -52,23 +53,6 @@ def test_command_mac_layout():
     assert sess.command_mac(0x3D, payload) == expected
 
 
-class QueueConn:
-    def __init__(self, responses):
-        self._responses = list(responses)
-        self.sent: list[bytes] = []
-
-    def transmit(self, apdu):
-        self.sent.append(bytes(apdu))
-        data, sw1, sw2 = self._responses.pop(0)
-        return list(data), sw1, sw2
-
-    def get_atr(self):
-        return b""
-
-    def disconnect(self):
-        pass
-
-
 class FakeEv2Card:
     """Implements the card side of AuthenticateEV2First for loopback testing."""
 

tests/unit/test_fido.py

@@ -2,6 +2,7 @@
 
 import cbor2
 import pytest
+from _cardfakes import QueueConn
 
 from cryptnox_id_cli.applets.fido import constants as c
 from cryptnox_id_cli.applets.fido.ctap import Ctap2Client, describe_get_info
@@ -14,23 +15,6 @@
 AAGUID = bytes.fromhex("1d1b4e3376a147fb97a014b10d0933f1")
 
 
-class QueueConn:
-    def __init__(self, responses):
-        self._responses = list(responses)
-        self.sent: list[bytes] = []
-
-    def transmit(self, apdu):
-        self.sent.append(bytes(apdu))
-        data, sw1, sw2 = self._responses.pop(0)
-        return list(data), sw1, sw2
-
-    def get_atr(self):
-        return b""
-
-    def disconnect(self):
-        pass
-
-
 def _info_payload() -> bytes:
     info = {
         c.INFO_VERSIONS: ["U2F_V2", "FIDO_2_0", "FIDO_2_1"],