Merge pull request #519 from LedgerHQ/support_nfc_apdu

add --nfc flag to use NFC apdu media

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+
+## [0.12.0] 2024-??-??
+
+### Added
+
+- NFC communication available
+- Starting Speculos with the `--transport` argument allows to choose U2F, HID or NFC transport
+- Flex and Stax OSes emulation always consider NFC to be up (it can't be deactivated for now)
+
 ## [0.11.0] 2024-11-12
 
 ### Added

speculos/main.py

@@ -29,6 +29,7 @@
 from .mcu.finger_tcp import FakeFinger
 from .mcu.struct import DisplayArgs, ServerArgs
 from .mcu.vnc import VNC
+from .mcu.transport import TransportType
 from .observer import BroadcastInterface
 from .resources_importer import resources
 
@@ -268,7 +269,9 @@ def main(prog=None) -> int:
                                                                    'to use a hex seed, prefix it with "hex:"')
     parser.add_argument('-t', '--trace', action='store_true', help='Trace syscalls')
     parser.add_argument('-u', '--usb', default='hid', help='Configure the USB transport protocol, '
-                                                           'either HID (default) or U2F')
+                        'either HID (default) or U2F (DEPRECATED, use `--transport` instead)')
+    parser.add_argument('-T', '--transport', default=None, choices=('HID', 'U2F', 'NFC'),
+                        help='Configure the transport protocol: HID (default), U2F or NFC.')
 
     group = parser.add_argument_group('network arguments')
     group.add_argument('--apdu-port', default=9999, type=int, help='ApduServer TCP port')
@@ -466,14 +469,20 @@ def main(prog=None) -> int:
     qemu_pid = run_qemu(s1, s2, args, use_bagl)
     s1.close()
 
+    # The `--transport` argument takes precedence over `--usb`
+    if args.transport is not None:
+        transport_type = TransportType[args.transport]
+    else:
+        transport_type = TransportType[args.usb.upper()]
+
     apdu = apdu_server.ApduServer(host="0.0.0.0", port=args.apdu_port)
     seph = seproxyhal.SeProxyHal(
         s2,
         model=args.model,
         use_bagl=use_bagl,
         automation=automation_path,
         automation_server=automation_server,
-        transport=args.usb)
+        transport=transport_type)
 
     button = None
     if args.button_port:

speculos/mcu/seproxyhal.py

@@ -8,7 +8,7 @@
 from typing import Callable, List, Optional, Tuple
 
 from speculos.observer import BroadcastInterface, TextEvent
-from . import usb
+from .transport import build_transport, TransportType
 from .automation import Automation
 from .display import DisplayNotifier, IODevice
 from .nbgl import NBGL
@@ -30,6 +30,9 @@ class SephTag(IntEnum):
     USB_CONFIG = 0x4f
     USB_EP_PREPARE = 0x50
 
+    NFC_RAPDU = 0x4A
+    NFC_POWER = 0x34
+
     REQUEST_STATUS = 0x52
     RAPDU = 0x53
     PLAY_TUNE = 0x56
@@ -253,7 +256,7 @@ def __init__(self,
                  use_bagl: bool,
                  automation: Optional[Automation] = None,
                  automation_server: Optional[BroadcastInterface] = None,
-                 transport: str = 'hid'):
+                 transport: TransportType = TransportType.HID):
         self._socket = sock
         self.logger = logging.getLogger("seproxyhal")
         self.printf_queue = ''
@@ -270,7 +273,7 @@ def __init__(self,
                                                    self.socket_helper.wait_until_tick_is_processed)
         self.time_ticker_thread.start()
 
-        self.usb = usb.USB(self.socket_helper.queue_packet, transport=transport)
+        self.transport = build_transport(self.socket_helper.queue_packet, transport)
 
         self.ocr = OCR(model, use_bagl)
 
@@ -389,10 +392,10 @@ def can_read(self, screen: DisplayNotifier):
                 c(data)
 
         elif tag == SephTag.USB_CONFIG:
-            self.usb.config(data)
+            self.transport.config(data)
 
         elif tag == SephTag.USB_EP_PREPARE:
-            data = self.usb.prepare(data)
+            data = self.transport.prepare(data)
             if data:
                 for c in self.apdu_callbacks:
                     c(data)
@@ -449,6 +452,13 @@ def can_read(self, screen: DisplayNotifier):
             assert isinstance(screen.display.gl, NBGL)
             screen.display.gl.hal_draw_image_file(data)
 
+        elif tag == SephTag.NFC_RAPDU:
+            data = self.transport.handle_rapdu(data)
+            if data is not None:
+                for c in self.apdu_callbacks:
+                    c(data)
+                screen.display.forward_to_apdu_client(data)
+
         else:
             self.logger.error(f"unknown tag: {tag:#x}")
             sys.exit(0)
@@ -506,7 +516,7 @@ def to_app(self, packet: bytes):
             tag, packet = packet[4], packet[5:]
             self.socket_helper.queue_packet(SephTag(tag), packet)
         else:
-            self.usb.xfer(packet)
+            self.transport.send(packet)
 
     def get_tick_count(self):
         return self.socket_helper.get_tick_count()

speculos/mcu/transport/__init__.py

@@ -0,0 +1,17 @@
+from typing import Callable
+
+from .interface import TransportLayer, TransportType
+from .nfc import NFC
+from .usb import HID, U2F
+
+
+def build_transport(cb: Callable, transport: TransportType) -> TransportLayer:
+    if transport is TransportType.NFC:
+        return NFC(cb, transport)
+    elif transport is TransportType.U2F:
+        return U2F(cb, transport)
+    else:
+        return HID(cb, transport)
+
+
+__all__ = ["build_transport", "TransportType"]

speculos/mcu/transport/interface.py

@@ -0,0 +1,36 @@
+from abc import ABC, abstractmethod
+from enum import auto, IntEnum
+from typing import Callable, Optional
+
+
+class TransportType(IntEnum):
+    HID = auto()
+    NFC = auto()
+    U2F = auto()
+
+
+class TransportLayer(ABC):
+
+    def __init__(self, send_cb: Callable, transport: TransportType):
+        self._transport = transport
+        self._send_cb = send_cb
+
+    @property
+    def type(self) -> TransportType:
+        return self._transport
+
+    @abstractmethod
+    def config(self, data: bytes) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def prepare(self, data: bytes) -> Optional[bytes]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def send(self, data: bytes) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def handle_rapdu(self, data: bytes) -> Optional[bytes]:
+        raise NotImplementedError

speculos/mcu/transport/nfc.py

@@ -0,0 +1,75 @@
+"""
+Forward NFC packets between the MCU and the SE
+"""
+
+import enum
+import logging
+from typing import List, Optional
+
+from .interface import TransportLayer, TransportType
+
+
+class SephNfcTag(enum.IntEnum):
+    NFC_APDU_EVENT = 0x1C
+    NFC_EVENT = 0x1E
+
+
+class NFC(TransportLayer):
+    def __init__(self, send_cb, transport: TransportType):
+        super().__init__(send_cb, transport)
+        self.MTU = 140
+        self.rx_sequence = 0
+        self.rx_size = 0
+        self.rx_data: bytes = b''
+        self.logger = logging.getLogger("NFC")
+
+    def config(self, data: bytes) -> None:
+        self.logger.warning("USB-specific 'config' method called on NFC transport. Ignored.")
+
+    def prepare(self, data: bytes) -> None:
+        self.logger.warning("USB-specific 'prepare' method called on NFC transport. Ignored.")
+
+    def handle_rapdu(self, data: bytes) -> Optional[bytes]:
+        """concatenate apdu chunks into full apdu"""
+        # example of data
+        # 0000050000002b3330000409312e302e302d72633104e600000008362e312e302d646508352e312e302d6465010001009000
+
+        # only APDU packets are supported
+        if data[2] != 0x05:
+            return None
+
+        sequence = int.from_bytes(data[3:5], 'big')
+        assert self.rx_sequence == sequence, f"Unexpected sequence number:{sequence}"
+
+        if sequence == 0:
+            self.rx_size = int.from_bytes(data[5:7], "big")
+            self.rx_data = data[7:]
+        else:
+            self.rx_data += data[5:]
+
+        if len(self.rx_data) == self.rx_size:
+            # prepare for next call
+            self.rx_sequence = 0
+            return self.rx_data
+        else:
+            self.rx_sequence += 1
+        return None
+
+    def send(self, data: bytes) -> None:
+        chunks: List[bytes] = []
+        data_len = len(data)
+
+        while len(data) > 0:
+            size = self.MTU - 5
+            chunks.append(data[:size])
+            data = data[size:]
+
+        for i, chunk in enumerate(chunks):
+            # Ledger protocol header
+            header = bytes([0x00, 0x00, 0x05])  # APDU
+            header += i.to_bytes(2, "big")
+            # first packet contains the size of full buffer
+            if i == 0:
+                header += data_len.to_bytes(2, "big")
+
+            self._send_cb(SephNfcTag.NFC_APDU_EVENT, header + chunk)