usb: add ncry v3 authenticated encryption

### docs/README.md
@@ -11,6 +11,7 @@ wants to understand why it's safe to put your moneys into Coldcard.
 - [`notes-on-repro.md`](notes-on-repro.md) Detailed breakdown of the reproducible build process.
 - [`upgrade-recovery.md`](upgrade-recovery.md) Firmware upgrade and recovery process.
 - [`backup-files.md`](backup-files.md) Some details of our encrypted backup files.
+- [`usb-ncry-v3.md`](usb-ncry-v3.md) Details of USB ncry v3 encrypted sessions and MITM-check best practices.
 - [`temporary-seeds.md`](temporary-seeds.md) Temporary (ephemeral) seeds and the Seed Vault.
 - [`seed-xor.md`](seed-xor.md) More about _Seed XOR_ feature, including fully worked Seed XOR example, and useful XOR lookup chart.
 - [`key-teleport.md`](key-teleport.md) Key Teleport: encrypted transfer of seeds and secrets between Q devices.
@@ -31,4 +32,3 @@ wants to understand why it's safe to put your moneys into Coldcard.
 - [`limitations.md`](limitations.md) Documented limitations, policy choices, and TODO items.
 - [`paperwallet.pdf`](paperwallet.pdf) Example paper wallet template file.
 - [`menu-tree.txt`](menu-tree.txt) Dump of the menu system. Incomplete, may be out of date.
-

### docs/usb-ncry-v3.md
@@ -0,0 +1,181 @@
+# USB ncry v3
+
+`ncry` is the USB command that starts encrypted communication between a
+host client and a COLDCARD. Earlier versions used ECDH to create an AES-CTR
+session stream. Version 3 keeps the same general setup flow, but changes the
+encrypted message format so each message is authenticated and each direction
+uses independent keys.
+
+## What v3 Does
+
+During setup, the host sends an ephemeral public key and requests
+`USB_NCRY_V3`. The device replies with its own ephemeral public key, master
+fingerprint, and master xpub information. Both sides use ECDH to compute the
+same session key.
+
+For v3, that session key is not used directly as one shared CTR stream. Instead
+the implementation runs it through an HMAC-SHA256 KDF and derives separate keys
+for:
+
+- host to device encryption
+- host to device message authentication
+- device to host encryption
+- device to host message authentication
+
+The KDF binds the output keys to the v3 protocol label, the requested version,
+and both ephemeral public keys:
+
+```
+transcript = SHA256("ccncry3" || version || host_pubkey || dev_pubkey)
+prk = HMAC-SHA256(key=transcript, message=session_key)
+okm = HKDF-Expand(prk, "ccncry3", 128)
+```
+
+Here `version` is the little-endian 32-bit `USB_NCRY_V3` value, and the public
+keys are the 64-byte uncompressed x/y values carried by `ncry`. The same short
+COLDCARD ncry v3 label is used for both the transcript hash and HKDF-Expand
+context.
+
+The 128-byte KDF output is split into four 32-byte keys, in the same order as
+the list above. Encrypted messages then use this wire format:
+
+```
+ciphertext = AES-256-CTR(plaintext)        # counter-0 stream, per direction
+tag        = HMAC-SHA256(key=mac_key, message=
+                 direction || LE32(sequence) || LE32(length) || ciphertext)[0:16]
+wire       = ciphertext || tag
+```
+
+The direction value is different for host-to-device and device-to-host traffic.
+The authentication tag is the first 16 bytes of the HMAC-SHA256 result. The
+sequence number is an unsigned little-endian 32-bit value. It starts at zero
+for each direction and increments after each valid message. Sequence number
+`0xffffffff` may be used once; the next message in that direction must fail
+instead of wrapping to zero.
+
+## Fixed Test Vector
+
+The following values are normative and are independently checked by the
+firmware and `ckcc-protocol` test suites:
+
+```text
+session_key = 000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f
+host_pubkey = 000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f
+              202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f
+device_pubkey = 404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f
+                606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f
+transcript = a5209d34d0dd4dc037e973145f388686e03a78c8e9db20e353a86022b19b1dce
+prk = ebf7de0dac1c1bd46a4290865e2fc49b9bdeb85cc66e5d73525c5f97602ab338
+h2d_enc = 4ae5bbe99e5565f58383634c28916469c6f9777392c5fc4d16e1a4ccffce51d4
+h2d_mac = 8641e0d7e0e9d7610cd33bc6c6025dd772faaad259bb6492ea59362ab7e284df
+d2h_enc = 176e6002488acf16c63646f483d7965c78b41159863e77710fd069a3c7b4ab32
+d2h_mac = 07bdfcd21c34b50241509b146675a5f06d385242dbaf4d0060e2ccfee819706e
+
+request plaintext = 70696e676e6372792d76332d766563746f72
+request wire = 4d90a086a4411368708106e3b6ae321fef9d
+               5fc1cfedaf2be32b1cdb55ce62b48029
+response plaintext = 62696e796e6372792d76332d766563746f72
+response wire = 026d1b5205e4686f3682acab956ac2690efe
+                f79887d299df4e655f887a112a7081d5
+```
+
+Both wire examples use sequence number zero. The request direction is
+`C2D\0`; the response direction is `D2C\0`.
+
+## Security Properties
+
+ncry v3 improves the USB encrypted channel in three broad areas:
+
+- Confidentiality: passive observers cannot read encrypted command or response
+  contents, and host-to-device and device-to-host traffic do not reuse the same
+  CTR stream.
+- Message integrity: tampered ciphertext is rejected before it is decrypted,
+  and CTR bit-flipping attacks are blocked by the message authentication tag.
+- Session ordering: same-session replay or reordering is rejected by the
+  sequence number, and cross-direction reflection is rejected by the direction
+  value in the MAC.
+
+After v3 setup, all future USB commands in that session are expected to be
+encrypted. Like v2, v3 is a bound USB mode: the device rejects a second `ncry`
+setup attempt for that USB handler.
+
+## What v3 Does Not Do
+
+ncry v3 does not authenticate the COLDCARD endpoint by itself. The ECDH public
+keys in the `ncry` setup are ephemeral and unsigned. An active attacker in the
+middle can establish one encrypted session with the host and another encrypted
+session with the device.
+
+In that attack, v3 still protects each individual encrypted session from
+passive reading, tampering, replay, and reflection. It does not prove that the
+host is talking directly to the intended COLDCARD. Endpoint authentication is
+provided by the separate `mitm` command.
+
+v3 also does not provide downgrade negotiation or recovery from a
+desynchronized stream. Any authentication or framing failure is terminal: the
+firmware stops processing USB commands for that session, and the client rejects
+further use of the connection. Reboot the Coldcard and reconnect.
+
+## Using v3 With MITM Check
+
+For sensitive clients, the recommended flow is:
+
+1. Open one `ColdcardDevice` connection with `ncry_ver=USB_NCRY_V3`.
+2. Immediately run `check_mitm(expected_xpub=trusted_xpub)`.
+3. Only continue with sensitive commands if the MITM check passes.
+4. If the MITM check or any v3 authentication check fails, close the session and
+   require a fresh connection.
+
+Example:
+
+```python
+from ckcc.client import ColdcardDevice
+from ckcc.constants import USB_NCRY_V3
+from ckcc.protocol import CCProtocolPacker
+
+trusted_xpub = "xpub..."
+
+dev = ColdcardDevice(ncry_ver=USB_NCRY_V3)
+try:
+    dev.check_mitm(expected_xpub=trusted_xpub)
+
+    # Run sensitive commands only after the endpoint check passes.
+    xpub = dev.send_recv(CCProtocolPacker.get_xpub("m"), timeout=None)
+finally:
+    dev.close()
+```
+
+The `expected_xpub` value should come from a trusted source, such as a previous
+trusted pairing, user verification, or another authenticated channel. If a
+client uses only the xpub returned by the same first-contact USB session, an
+active attacker could substitute its own identity and pass the check against
+that substituted value. The MITM check is strongest when the host already knows
+which COLDCARD it expects.
+
+The MITM command works by asking the COLDCARD to sign the current session key
+with its master key. The host verifies that signature against the trusted xpub.
+Because an active MITM creates a different session key on each side, it cannot
+forward the real device's signature and make it verify for the host's separate
+session.
+
+## Operational Notes
+
+First supported in:
+
+- ckcc-protocol client: `1.6.0`
+- COLDCARD Mk4/Mk5 firmware: `5.6.1`
+- COLDCARD Q firmware: `1.5.1Q`
+
+Use v3 only when both the client and firmware are known to support it. The
+default client encryption version remains `USB_NCRY_V1` for compatibility, so a
+client must explicitly opt in to v3.
+
+The ncry version is selected by the host in the initial `ncry` command. There
+is no in-protocol version negotiation or downgrade path. If firmware does not
+support `USB_NCRY_V3`, it rejects the setup request and the client must close
+that attempt.
+
+Use one client-side request/response flow at a time. Do not pipeline or
+interleave commands in a v3 session. The sequence numbers and CTR streams are
+stateful per direction, so the next command should be sent only after the
+previous response has been received and authenticated.

### external/ckcc-protocol
@@ -1 +1 @@
-Subproject commit 3d1dfa858beb58b8dac37d8c66d7aed2909812f2
+Subproject commit 2d4b22a1fc9c9449e3b3b434958b978c683a3a6c

### releases/Next-ChangeLog.md
@@ -4,6 +4,7 @@ This lists the new changes that have not yet been published in a normal release.
 
 # Shared Improvements - Both Mk and Q
 
+- New Feature: Added USB ncry v3 authenticated encryption with direction-separated keys and replay protection
 - Enhancement: Warn when a transaction's block-height `nLockTime` is more than
   ten years beyond the Bitcoin block height known to the firmware.
 - Bugfix: Reject duplicate singleton keys in PSBT maps

### shared/usb.py
@@ -5,11 +5,14 @@
 import ckcc, pyb, callgate, sys, ux, ngu, stash, aes256ctr
 from uasyncio import sleep_ms, core
 from uhashlib import sha256
-from public_constants import MAX_MSG_LEN, MAX_BLK_LEN, AFC_SCRIPT
-from public_constants import STXN_FLAGS_MASK
+from public_constants import (
+    AFC_SCRIPT, MAX_BLK_LEN, MAX_MSG_LEN, STXN_FLAGS_MASK,
+    USB_V3_C2D, USB_V3_D2C, USB_V3_KDF_LABEL,
+    USB_V3_MAX_WIRE_MSG_LEN, USB_V3_TAG_LEN,
+)
 from ustruct import pack, unpack_from
 from ckcc import watchpoint, is_simulator
-from utils import problem_file_line, call_later_ms, to_ascii_printable
+from utils import problem_file_line, call_later_ms, consteq, hkdf_expand, to_ascii_printable
 from version import supports_hsm, is_devmode, MAX_TXN_LEN, MAX_UPLOAD_LEN
 from exceptions import FramingError, CCBusyError, HSMDenied, HSMCMDDisabled, SpendPolicyViolation
 from pincodes import pa
@@ -77,6 +80,32 @@
     'bagi', 'dfu_',     # just in case
 }) | HSM_DISABLE_CMDS
 
+def usb_v3_keys(session_key, host_pubkey, dev_pubkey):
+    # Bind derived keys to v3 and to both ephemeral public keys. This prevents
+    # key reuse across directions and across any future ncry transcript shape.
+    transcript = ngu.hash.sha256s(
+        USB_V3_KDF_LABEL +
+        pack('<I', 0x3) +
+        host_pubkey +
+        dev_pubkey
+    )
+
+    prk = ngu.hmac.hmac_sha256(transcript, session_key)
+    okm = hkdf_expand(prk, USB_V3_KDF_LABEL, 128)
+
+    return (
+        okm[0:32],       # host -> device AES-CTR key
+        okm[32:64],      # host -> device HMAC key
+        okm[64:96],      # device -> host AES-CTR key
+        okm[96:128],     # device -> host HMAC key
+    )
+
+
+def usb_v3_tag(mac_key, direction, seq, ciphertext):
+    mac_msg = pack('<4sII', direction, seq, len(ciphertext)) + bytes(ciphertext)
+    return ngu.hmac.hmac_sha256(mac_key, mac_msg)[0:USB_V3_TAG_LEN]
+
+
 # singleton instance of USBHandler()
 handler = None
 
@@ -137,8 +166,8 @@ def __init__(self):
         # handle simulator
         self.blockable = getattr(self.dev, 'pipe', self.dev)
 
-        self.msg = bytearray(2048+12)
-        assert len(self.msg) == MAX_MSG_LEN
+        self.msg = bytearray(USB_V3_MAX_WIRE_MSG_LEN)
+        assert len(self.msg) == USB_V3_MAX_WIRE_MSG_LEN
 
         self.encrypted_req = False
 
@@ -148,6 +177,12 @@ def __init__(self):
         # these will be objects later
         self.encrypt = None
         self.decrypt = None
+        self.ncry_ver = 0
+        self.rx_mac_key = None
+        self.tx_mac_key = None
+        self.rx_seq = 0
+        self.tx_seq = 0
+        self.v3_failed = False
 
     def get_packet(self):
         # read next packet (64 bytes) waiting on the wire. Unframe it and return
@@ -177,6 +212,9 @@ async def usb_hid_recv(self):
         msg_len = 0
 
         while 1:
+            if self.v3_failed:
+                return
+
             success = False
             yield core._io_queue.queue_read(self.blockable)
 
@@ -186,7 +224,9 @@ async def usb_hid_recv(self):
                 #print('Rx[%d]' % len(here))
                 if here:
                     lh = len(here)
-                    if msg_len+lh > MAX_MSG_LEN:
+                    max_wire_len = (USB_V3_MAX_WIRE_MSG_LEN
+                                    if self.ncry_ver == 0x3 else MAX_MSG_LEN)
+                    if msg_len+lh > max_wire_len:
                         raise FramingError('xlong')
 
                     self.msg[msg_len:msg_len + lh] = here
@@ -202,7 +242,10 @@ async def usb_hid_recv(self):
                     # need more content
                     continue
 
-                if not(4 <= msg_len <= MAX_MSG_LEN):
+                max_wire_len = (USB_V3_MAX_WIRE_MSG_LEN
+                                if self.ncry_ver == 0x3 and is_encrypted
+                                else MAX_MSG_LEN)
+                if not(4 <= msg_len <= max_wire_len):
                     raise FramingError('badsz')
 
                 if not is_encrypted and self.bound:
@@ -213,7 +256,7 @@ async def usb_hid_recv(self):
                         raise FramingError('no key')
 
                     self.encrypted_req = True
-                    self.decrypt_inplace(msg_len)
+                    msg_len = self.decrypt_inplace(msg_len)
                 else:
                     self.encrypted_req = False
 
@@ -268,9 +311,24 @@ async def usb_hid_recv(self):
                 reason = exc.args[0]
                 # print("Framing: %s" % reason)
                 self.reset_upload()
-                await self.framing_error(reason)
+                if self.ncry_ver == 0x3:
+                    try:
+                        await self.framing_error(reason)
+                    except FramingError:
+                        # A v3 error response can fail when tx sequence
+                        # is exhausted. The v3 session is terminal anyway.
+                        pass
+                else:
+                    await self.framing_error(reason)
                 msg_len = 0
 
+                # Authentication/framing failure consumes or invalidates the
+                # peer's stateful v3 stream. Never process another command in
+                # this session; a reboot is required to create a fresh one.
+                if self.ncry_ver == 0x3:
+                    self.v3_failed = True
+                    return
+
             except BaseException as exc:
                 # recover from general issues/keep going
                 #print("USB!")
@@ -288,10 +346,43 @@ def decrypt_inplace(self, msg_len):
         # self.msg is encrypted. decode it in place
         # - seems dangerous to use memview here, but works
         # - some memory alloc still happens here tho
+        if self.ncry_ver == 0x3:
+            if self.rx_seq > 0xffffffff:
+                raise FramingError('seq')
+
+            if msg_len <= USB_V3_TAG_LEN:
+                raise FramingError('auth')
+
+            ct_len = msg_len - USB_V3_TAG_LEN
+            if ct_len < 4:
+                raise FramingError('badsz')
+
+            ciphertext = memoryview(self.msg)[0:ct_len]
+            got_tag = memoryview(self.msg)[ct_len:msg_len]
+
+            expect_tag = usb_v3_tag(
+                self.rx_mac_key, USB_V3_C2D, self.rx_seq, ciphertext)
+            if not consteq(got_tag, expect_tag):
+                raise FramingError('auth')
+
+            self.msg[0:ct_len] = self.decrypt(ciphertext)
+            self.rx_seq += 1
+            return ct_len
+
         self.msg[0:msg_len] = self.decrypt(memoryview(self.msg)[0:msg_len])
+        return msg_len
 
     def encrypt_response(self, msg):
         # encrypt what we'll send to desktop
+        if self.ncry_ver == 0x3:
+            if self.tx_seq > 0xffffffff:
+                raise FramingError('seq')
+
+            ciphertext = self.encrypt(msg)
+            tag = usb_v3_tag(
+                self.tx_mac_key, USB_V3_D2C, self.tx_seq, ciphertext)
+            self.tx_seq += 1
+            return bytes(ciphertext) + tag
 
         return self.encrypt(msg)
 
@@ -717,13 +808,13 @@ def call_after(self, func, *args):
     def handle_crypto_setup(self, version, his_pubkey):
         # pick a one-time key pair for myself, and return the pubkey for that
         # determine what the session key will be for this connection
-        if version not in [0x1, 0x2]:
+        if version not in [0x1, 0x2, 0x3]:
             raise FramingError('bad ncry version')
         assert len(his_pubkey) == 64
 
         if self.bound:
             raise FramingError('crypto already set up')
-        if version == 0x2:
+        if version in [0x2, 0x3]:
             self.bound = True
 
         # new session: any download lease from a previous session is void
@@ -740,13 +831,36 @@ def handle_crypto_setup(self, version, his_pubkey):
         self.session_key = pair.ecdh_multiply(b'\x04' + his_pubkey)
         del pair
 
-        #print("session = " + str(b2a_hex(self.session_key)))
-
-        # Would be nice to have nonce in addition to the counter, but
-        # harder on the desktop side.
-        ctr = aes256ctr.new(self.session_key)
-        self.encrypt = ctr.cipher
-        self.decrypt = ctr.copy().cipher
+        self.ncry_ver = version
+        self.rx_seq = 0
+        self.tx_seq = 0
+        self.v3_failed = False
+        self.rx_mac_key = None
+        self.tx_mac_key = None
+
+        if version == 0x3:
+            # ncry v3 derives independent keys for each direction
+            # and authenticates every encrypted message before decrypting it:
+            #
+            #   wire = AES-CTR(plaintext) || HMAC(direction, seq, len, ciphertext)
+            #
+            # The sequence number makes replay/reordering detectable within a
+            # session, the direction label prevents cross-direction reflection,
+            # and the MAC removes CTR's normal bit-flipping malleability.
+            h2d_enc, h2d_mac, d2h_enc, d2h_mac = usb_v3_keys(
+                self.session_key,
+                his_pubkey,
+                my_pubkey[1:],
+            )
+            self.decrypt = aes256ctr.new(h2d_enc).cipher
+            self.encrypt = aes256ctr.new(d2h_enc).cipher
+            self.rx_mac_key = h2d_mac
+            self.tx_mac_key = d2h_mac
+        else:
+            # v1/v2 legacy wire format. Kept unchanged for compatibility.
+            ctr = aes256ctr.new(self.session_key)
+            self.encrypt = ctr.cipher
+            self.decrypt = ctr.copy().cipher
 
         from glob import settings
         xfp = settings.get('xfp', 0)

### shared/users.py
@@ -11,14 +11,13 @@
 from menu import MenuSystem, MenuItem
 from ucollections import namedtuple
 from ux import ux_dramatic_pause, ux_show_story, ux_confirm
+from utils import consteq
 from glob import settings
 
 # accepting strings and strings, returning bytes when decoding, str when encoding (ie. correct)
 b32encode = ngu.codecs.b32_encode
 b32decode = ngu.codecs.b32_decode
 
-hmac_sha256 = ngu.hmac.hmac_sha256
-
 # to keep menus and such to a reasonable size
 MAX_NUMBER_USERS = const(30)
 
@@ -65,7 +64,7 @@ def calc_local_pincode(psbt_sha, hmac_secret):
     from ubinascii import a2b_base64
     key = a2b_base64(hmac_secret)
     assert len(psbt_sha) == 32
-    digest = hmac_sha256(key, psbt_sha)
+    digest = ngu.hmac.hmac_sha256(key, psbt_sha)
 
     num = ustruct.unpack('>I', digest[-4:])[0] & 0x7fffffff
     return '%06d' % (num % 1000000)
@@ -201,8 +200,8 @@ def auth_okay(cls, username, token, totp_time=None, psbt_hash=None):
         secret = b32decode(secret)
 
         if auth_mode == USER_AUTH_HMAC:
-            expect = hmac_sha256(secret, psbt_hash or bytes(32))
-            if expect != token:
+            expect = ngu.hmac.hmac_sha256(secret, psbt_hash or bytes(32))
+            if not consteq(expect, token):
                 return 'mismatch'
 
             if last_counter == 0:
@@ -240,7 +239,7 @@ def auth_okay(cls, username, token, totp_time=None, psbt_hash=None):
 
             #print('expect=%r got=%r cnt=%d last=%d' % (expect, token, c, last_counter))
 
-            if expect == token:
+            if consteq(expect, token):
                 # success, need to update last counter level seen (especially for HOTP,
                 # but also to resist replay for TOTP)
                 cls.update_counter(username, c)

### shared/utils.py
@@ -90,6 +90,29 @@ def pop_count(i):
 
     return (((i + (i >> 4) & 0xF0F0F0F) * 0x1010101) & 0xffffffff) >> 24
 
+def consteq(a, b):
+    # Constant-time compare for equal-length byte strings.
+    if len(a) != len(b):
+        return False
+
+    diff = 0
+    for idx in range(len(a)):
+        diff |= a[idx] ^ b[idx]
+    return diff == 0
+
+def hkdf_expand(prk, info, length):
+    # RFC5869 HKDF-Expand with HMAC-SHA256.
+    out = bytearray()
+    t = b''
+    counter = 1
+
+    while len(out) < length:
+        t = ngu.hmac.hmac_sha256(prk, t + info + bytes([counter]))
+        out.extend(t)
+        counter += 1
+
+    return bytes(out[0:length])
+
 def get_filesize(fn):
     # like os.path.getsize()
     try:

### testing/devtest/unit_ncry_v3.py
@@ -0,0 +1,34 @@
+import aes256ctr
+from ubinascii import unhexlify as a2b_hex
+
+from public_constants import USB_V3_C2D, USB_V3_D2C
+from usb import usb_v3_keys, usb_v3_tag
+
+
+session_key = bytes(range(32))
+host_pubkey = bytes(range(64))
+dev_pubkey = bytes(range(64, 128))
+
+expect_keys = (
+    a2b_hex(b'4ae5bbe99e5565f58383634c28916469c6f9777392c5fc4d16e1a4ccffce51d4'),
+    a2b_hex(b'8641e0d7e0e9d7610cd33bc6c6025dd772faaad259bb6492ea59362ab7e284df'),
+    a2b_hex(b'176e6002488acf16c63646f483d7965c78b41159863e77710fd069a3c7b4ab32'),
+    a2b_hex(b'07bdfcd21c34b50241509b146675a5f06d385242dbaf4d0060e2ccfee819706e'),
+)
+
+keys = usb_v3_keys(session_key, host_pubkey, dev_pubkey)
+assert keys == expect_keys
+
+request_plaintext = a2b_hex(b'70696e676e6372792d76332d766563746f72')
+request_ciphertext = aes256ctr.new(keys[0]).cipher(request_plaintext)
+assert request_ciphertext == a2b_hex(
+    b'4d90a086a4411368708106e3b6ae321fef9d')
+assert usb_v3_tag(keys[1], USB_V3_C2D, 0, request_ciphertext) == a2b_hex(
+    b'5fc1cfedaf2be32b1cdb55ce62b48029')
+
+response_plaintext = a2b_hex(b'62696e796e6372792d76332d766563746f72')
+response_ciphertext = aes256ctr.new(keys[2]).cipher(response_plaintext)
+assert response_ciphertext == a2b_hex(
+    b'026d1b5205e4686f3682acab956ac2690efe')
+assert usb_v3_tag(keys[3], USB_V3_D2C, 0, response_ciphertext) == a2b_hex(
+    b'f79887d299df4e655f887a112a7081d5')

### testing/ncry_tests.py
@@ -0,0 +1,300 @@
+# (c) Copyright 2026 by Coinkite Inc. This file is covered by license found in COPYING-CC.
+#
+import pytest
+from hashlib import sha256
+
+from ckcc_protocol.client import ColdcardDevice
+from ckcc_protocol.constants import USB_NCRY_V1, USB_NCRY_V2, USB_NCRY_V3, USB_V3_TAG_LEN
+from ckcc_protocol.protocol import (
+    MAX_MSG_LEN,
+    CCFramingError,
+    CCProtocolPacker,
+    CCProtocolUnpacker,
+)
+from run_sim_tests import ColdcardSimulator, clean_sim_data, remove_all_client_sockets
+
+
+def xor_bytes(left, right):
+    assert len(left) == len(right)
+    return bytes(a ^ b for a, b in zip(left, right))
+
+
+@pytest.fixture
+def ncry_v3_dev(request):
+    clean_sim_data()
+    remove_all_client_sockets()
+
+    sim_args = ["--eff", "--set", "nfc=1"]
+    if request.config.getoption("--Q"):
+        sim_args.append("--q1")
+
+    sim = ColdcardSimulator(
+        args=sim_args,
+        headless=request.config.getoption("--headless"))
+    dev = None
+
+    try:
+        sim.start(start_wait=3)
+        dev = ColdcardDevice(
+            sn=sim.socket,
+            is_simulator=True,
+            ncry_ver=USB_NCRY_V3)
+
+        yield dev
+    finally:
+        if dev is not None:
+            dev.close()
+        if sim.proc is not None:
+            sim.stop()
+        clean_sim_data()
+        remove_all_client_sockets()
+
+
+def send_wire(dev, wire, encrypted=True, timeout=3000):
+    left = len(wire)
+    offset = 0
+    while left > 0:
+        here = min(63, left)
+        buf = bytearray(65)
+        buf[2:2+here] = wire[offset:offset+here]
+        if here == left:
+            buf[1] = here | 0x80 | (0x40 if encrypted else 0x00)
+        else:
+            buf[1] = here
+
+        assert dev.dev.write(buf) == len(buf)
+        offset += here
+        left -= here
+
+    resp = b''
+    while True:
+        buf = dev.dev.read(64, timeout_ms=timeout)
+        assert buf, "timeout reading USB EP"
+
+        flag = buf[0]
+        resp += bytes(buf[1:1+(flag & 0x3f)])
+        if flag & 0x80:
+            return flag, resp
+
+
+def send_encrypted(dev, msg):
+    wire = dev.encrypt_request(msg)
+    flag, resp = send_wire(dev, wire, encrypted=True)
+    assert flag & 0x40
+    return wire, resp
+
+
+def decode_encrypted_response(dev, wire):
+    return CCProtocolUnpacker.decode(dev.decrypt_response(wire))
+
+
+def assert_encrypted_framing_error(dev, wire, reason):
+    plaintext = dev.decrypt_response(wire)
+    assert plaintext == b'fram' + reason.encode()
+    with pytest.raises(CCFramingError, match=reason):
+        CCProtocolUnpacker.decode(plaintext)
+
+
+def test_ncry_v3_single_client_multiple_commands(ncry_v3_dev):
+    dev = ncry_v3_dev
+
+    assert dev.ncry_ver == USB_NCRY_V3
+    assert dev.session_key
+
+    rb = dev.send_recv(CCProtocolPacker.ping(b'\x5a' * 32))
+    assert rb == b'\x5a' * 32
+
+    version = dev.send_recv(CCProtocolPacker.version())
+    assert '\n' in version
+
+    chain = dev.send_recv(CCProtocolPacker.block_chain())
+    assert chain in {'BTC', 'XTN', 'XRT'}
+
+    xpub = dev.send_recv(CCProtocolPacker.get_xpub('m'), timeout=None)
+    assert xpub[1:4] == 'pub'
+
+    data = b'ncry-v3-single-client'
+    assert dev.send_recv(CCProtocolPacker.upload(0, len(data), data)) == 0
+    assert dev.send_recv(CCProtocolPacker.sha256()) == sha256(data).digest()
+
+    rb = dev.send_recv(CCProtocolPacker.ping(bytes(MAX_MSG_LEN-4)))
+    assert set(rb) == {0} and len(rb) == MAX_MSG_LEN-4
+
+
+def test_ncry_v3_fixed_vectors(ncry_v3_dev, src_root_dir):
+    hook = 'execfile("%s/testing/devtest/unit_ncry_v3.py")' % src_root_dir
+    assert ncry_v3_dev.send_recv(b'EXEC' + hook.encode()) == b''
+
+
+def test_ncry_v3_overlong_response_rejected(ncry_v3_dev):
+    # EXEC adds the four-byte "biny" response prefix. Exceed the maximum
+    # authenticated v3 wire response by exactly one byte.
+    cmd = "RV.write(b'x' * %d)" % (MAX_MSG_LEN - 3)
+
+    with pytest.raises(CCFramingError, match="Response too long"):
+        ncry_v3_dev.send_recv(b'EXEC' + cmd.encode())
+    assert ncry_v3_dev._v3_failed
+
+
+def test_ncry_v3_directional_streams_cannot_be_xored_to_decrypt(ncry_v3_dev):
+    dev = ncry_v3_dev
+    payload = b'\x33' * 32
+    request_plaintext = CCProtocolPacker.ping(payload)
+    response_plaintext = b'biny' + payload
+
+    request_wire, response_wire = send_encrypted(dev, request_plaintext)
+    request_ciphertext = request_wire[:-USB_V3_TAG_LEN]
+    response_ciphertext = response_wire[:-USB_V3_TAG_LEN]
+
+    assert xor_bytes(request_ciphertext, response_ciphertext) != xor_bytes(
+        request_plaintext, response_plaintext)
+    recovered_response_plaintext = xor_bytes(
+        response_ciphertext,
+        xor_bytes(request_ciphertext, request_plaintext))
+    assert recovered_response_plaintext != response_plaintext
+    assert decode_encrypted_response(dev, response_wire) == payload
+
+
+def test_ncry_v3_response_replay_rejected(ncry_v3_dev):
+    dev = ncry_v3_dev
+    payload = b'\x44' * 16
+
+    _, response_wire = send_encrypted(dev, CCProtocolPacker.ping(payload))
+    assert decode_encrypted_response(dev, response_wire) == payload
+
+    with pytest.raises(CCFramingError):
+        dev.decrypt_response(response_wire)
+
+
+def test_ncry_v3_response_tamper_rejected_before_sequence_increment(ncry_v3_dev):
+    dev = ncry_v3_dev
+    payload = b'\x55' * 16
+
+    _, response_wire = send_encrypted(dev, CCProtocolPacker.ping(payload))
+    tampered_response = bytearray(response_wire)
+    tampered_response[0] ^= 1
+
+    rx_seq = dev.rx_seq
+    with pytest.raises(CCFramingError):
+        dev.decrypt_response(bytes(tampered_response))
+    assert dev.rx_seq == rx_seq
+    assert dev._v3_failed
+    with pytest.raises(CCFramingError, match='session failed'):
+        dev.decrypt_response(response_wire)
+
+
+def test_ncry_v3_request_mac_auth_failure_rejected(ncry_v3_dev):
+    dev = ncry_v3_dev
+    bad_request = bytearray(dev.encrypt_request(
+        CCProtocolPacker.ping(b'\x66' * 16)))
+    bad_request[-1] ^= 1
+
+    flag, response_wire = send_wire(dev, bytes(bad_request), encrypted=True)
+    assert flag & 0x40
+    assert_encrypted_framing_error(dev, response_wire, 'auth')
+
+    # Firmware terminates its USB receive task after the authenticated error.
+    next_request = dev.encrypt_request(CCProtocolPacker.ping(b'after-auth-failure'))
+    with pytest.raises(AssertionError, match='timeout reading USB EP'):
+        send_wire(dev, next_request, encrypted=True, timeout=100)
+
+
+def test_ncry_v3_request_replay_rejected(ncry_v3_dev):
+    dev = ncry_v3_dev
+    request_wire, response_wire = send_encrypted(
+        dev, CCProtocolPacker.ping(b'\x77' * 16))
+    assert decode_encrypted_response(dev, response_wire) == b'\x77' * 16
+
+    flag, replay_response_wire = send_wire(dev, request_wire, encrypted=True)
+    assert flag & 0x40
+    assert_encrypted_framing_error(dev, replay_response_wire, 'auth')
+
+
+def test_ncry_v3_short_encrypted_request_rejected(ncry_v3_dev):
+    dev = ncry_v3_dev
+
+    flag, response_wire = send_wire(dev, bytes(USB_V3_TAG_LEN), encrypted=True)
+    assert flag & 0x40
+    assert_encrypted_framing_error(dev, response_wire, 'auth')
+
+
+def test_ncry_v3_short_ciphertext_request_rejected(ncry_v3_dev):
+    dev = ncry_v3_dev
+
+    flag, response_wire = send_wire(
+        dev, bytes(USB_V3_TAG_LEN + 1), encrypted=True)
+    assert flag & 0x40
+    assert_encrypted_framing_error(dev, response_wire, 'badsz')
+@pytest.fixture
+def ncry_legacy_dev(request):
+    # Same simulator boot as ncry_v3_dev, but with a legacy (v1/v2) session.
+    clean_sim_data()
+    remove_all_client_sockets()
+
+    sim_args = ["--eff", "--set", "nfc=1"]
+    if request.config.getoption("--Q"):
+        sim_args.append("--q1")
+
+    sim = ColdcardSimulator(
+        args=sim_args,
+        headless=request.config.getoption("--headless"))
+    dev = None
+
+    try:
+        sim.start(start_wait=3)
+        dev = ColdcardDevice(
+            sn=sim.socket,
+            is_simulator=True,
+            ncry_ver=request.param)
+
+        yield dev
+    finally:
+        if dev is not None:
+            dev.close()
+        if sim.proc is not None:
+            sim.stop()
+        clean_sim_data()
+        remove_all_client_sockets()
+
+
+def test_ncry_legacy_wire_format_unchanged():
+    # Legacy v1/v2 wire format must be exactly the pre-v3 format:
+    # bare AES-CTR keystream output, no authentication tag appended.
+    # (pure client-side; calling encrypt_request on a live session would
+    # advance the CTR stream and desync the link)
+    import pyaes
+
+    session_key = sha256(b'ncry-legacy-format').digest()
+    msg = CCProtocolPacker.ping(b'\x5a' * 32)
+
+    dev = ColdcardDevice.__new__(ColdcardDevice)
+    dev.aes_setup(session_key)
+
+    wire = dev.encrypt_request(msg)
+    assert len(wire) == len(msg), "legacy wire format must not carry a tag"
+
+    # device side decrypts with an independent counter-0 CTR on the same key
+    dev_ctr = pyaes.AESModeOfOperationCTR(session_key, pyaes.Counter(0))
+    assert dev_ctr.decrypt(wire) == msg
+
+
+@pytest.mark.parametrize('ncry_legacy_dev', [USB_NCRY_V1, USB_NCRY_V2], indirect=True)
+def test_ncry_legacy_encryption_unchanged(ncry_legacy_dev):
+    # Legacy v1/v2 sessions must keep working end-to-end against firmware
+    # that also supports ncry v3.
+    dev = ncry_legacy_dev
+
+    assert dev.ncry_ver in (USB_NCRY_V1, USB_NCRY_V2)
+
+    rb = dev.send_recv(CCProtocolPacker.ping(b'\x5a' * 32))
+    assert rb == b'\x5a' * 32
+
+    version = dev.send_recv(CCProtocolPacker.version())
+    assert '\n' in version
+
+    xpub = dev.send_recv(CCProtocolPacker.get_xpub('m'), timeout=None)
+    assert xpub[1:4] == 'pub'
+
+    data = b'ncry-legacy-compat'
+    assert dev.send_recv(CCProtocolPacker.upload(0, len(data), data)) == 0
+    assert dev.send_recv(CCProtocolPacker.sha256()) == sha256(data).digest()

### testing/run_sim_tests.py
@@ -10,6 +10,7 @@
 python run_sim_tests.py --veryslow                             # run ONLY very slow tests
 python run_sim_tests.py --onetime                              # run ONLY onetime tests (each will get its own simulator)
 python run_sim_tests.py --onetime --veryslow                   # run both onetime and very slow
+python run_sim_tests.py --ncry3                                # run ncry v3 simulator/client tests
 python run_sim_tests.py -m test_nfc.py                         # run only nfc tests
 python run_sim_tests.py -m test_nfc.py -m test_hsm.py          # run nfc and hsm tests
 python run_sim_tests.py -m all                                 # run all tests but not onetime and not very slow (cca 40 minutes)
@@ -183,7 +184,7 @@ def _run_pytest_tests(test_module: str, pytest_marks: str, pytest_k: str, pdb: b
     if psbt2:
         cmd_list.append("--psbt2")
     if is_Q:
-        cmd_list.insert(0, "--Q")  # only changes behavior in login_settings_test
+        cmd_list.insert(0, "--Q")
     if headless:
         cmd_list.append("--headless")
     if sim_socket:
@@ -312,6 +313,8 @@ def main():
                         help="run 'clone_tests'")
     parser.add_argument("--seedless", action="store_true", default=False,
                         help="run 'seedless_tests'")
+    parser.add_argument("--ncry3", action="store_true", default=False,
+                        help="run 'ncry_tests'")
     parser.add_argument("--collect", type=str, metavar="MARK",
                         help="Collect marked test and print them to stdout")
     parser.add_argument("-k", "--pytest-k", type=str, metavar="EXPRESSION", default=None,
@@ -339,7 +342,8 @@ def main():
                                 and args.veryslow is False
                                 and args.login is False
                                 and args.clone is False
-                                and args.seedless is False):
+                                and args.seedless is False
+                                and args.ncry3 is False):
         args.module = ["all"]
 
     DEFAULT_SIMULATOR_ARGS = ["--eff", "--set", "nfc=1"]
@@ -581,9 +585,18 @@ def add_to_queue(module_name, simulator_args, queue):
         print("start seedless tests")
         ec, failed_tests = run_coldcard_tests(test_module="seedless_tests.py", pdb=args.pdb,
                                               failed_first=args.ff, pytest_k=args.pytest_k,
+                                              is_Q=True if args.q1 else False,
                                               headless=args.headless)
         result.append((f"seedless_tests", ec, failed_tests))
 
+    if args.ncry3:
+        print("start ncry3 tests")
+        ec, failed_tests = run_coldcard_tests(test_module="ncry_tests.py", pdb=args.pdb,
+                                              failed_first=args.ff, pytest_k=args.pytest_k,
+                                              is_Q=True if args.q1 else False,
+                                              headless=args.headless)
+        result.append((f"ncry_tests", ec, failed_tests))
+
     print("All done")
 
     any_failed = False

### testing/seedless_tests.py
@@ -1,48 +1,52 @@
 # (c) Copyright 2024 by Coinkite Inc. This file is covered by license found in COPYING-CC.
 #
 import pytest, pdb, time, random, os, shutil
-from charcodes import KEY_CANCEL, KEY_QR
+from charcodes import KEY_QR
 from core_fixtures import _pick_menu_item, _press_select, _press_cancel, _word_menu_entry
 from core_fixtures import _need_keypress, _sim_exec, _cap_story
 from run_sim_tests import ColdcardSimulator, clean_sim_data
 from ckcc_protocol.client import ColdcardDevice
 from ckcc_protocol.protocol import CCProtocolPacker
 
 
-def test_status_bar_rewrite_after_restore_master():
+def test_status_bar_rewrite_after_restore_master(request):
     from PIL import Image
+    is_Q = request.config.getoption('--Q')
     clean_sim_data()  # remove all from previous
-    sim = ColdcardSimulator(args=["--q1", "-l"])
+    sim_args = ["-l"]
+    if is_Q:
+        sim_args.append("--q1")
+    sim = ColdcardSimulator(args=sim_args)
     sim.start(start_wait=3)
     device = ColdcardDevice(is_simulator=True)
 
-    _pick_menu_item(device, True, "Advanced/Tools")
-    _pick_menu_item(device, True, "Temporary Seed")
+    _pick_menu_item(device, is_Q, "Advanced/Tools")
+    _pick_menu_item(device, is_Q, "Temporary Seed")
     _need_keypress(device, "4")
-    _pick_menu_item(device, True, "Generate Words")
-    _pick_menu_item(device, True, "12 Words")
+    _pick_menu_item(device, is_Q, "Generate Words")
+    _pick_menu_item(device, is_Q, "12 Words")
 
-    _pick_menu_item(device, True, "Mash Keys")
+    _pick_menu_item(device, is_Q, "Mash Keys")
     time.sleep(.1)
-    _press_select(device, True)
+    _press_select(device, is_Q)
 
     for i in range(65):
         _need_keypress(device, str(i % 10))
 
     time.sleep(.2)
-    _press_select(device, True)
+    _press_select(device, is_Q)
 
     _need_keypress(device, "6")
-    _press_select(device, True)
-    _press_select(device, True)
-    _need_keypress(device, KEY_CANCEL)
-    _need_keypress(device, KEY_CANCEL)
+    _press_select(device, is_Q)
+    _press_select(device, is_Q)
+    _press_cancel(device, is_Q)
+    _press_cancel(device, is_Q)
     fn0 = os.path.realpath(f'./debug/seedless-status-snap-{random.randint(int(1E6), int(9E6))}.png')
     _sim_exec(device, f"from glob import dis; dis.dis.save_snapshot({fn0!r})")
     time.sleep(1)
     rv0 = Image.open(fn0)
-    _pick_menu_item(device, True, "Restore Master")
-    _press_select(device, True)
+    _pick_menu_item(device, is_Q, "Restore Master")
+    _press_select(device, is_Q)
     fn1 = os.path.realpath(f'./debug/seedless-status-snap-{random.randint(int(1E6), int(9E6))}.png')
     _sim_exec(device, f"from glob import dis; dis.dis.save_snapshot({fn1!r})")
     time.sleep(1)