release: python-keepkey 7.14.0 — all chain support + CI + zoo

.github/workflows/ci.yml

@@ -0,0 +1,162 @@
+# KeepKey python-keepkey CI
+#
+# Pulls the published emulator image (kktech/kkemu) from DockerHub
+# and runs the full python integration test suite against it.
+#
+# Stage 1: GATE (seconds)
+#   └─ lint        basic Python syntax check
+#
+# Stage 2: TEST (gated by Stage 1)
+#   └─ integration full pytest suite against emulator
+
+name: CI
+
+on:
+  push:
+    branches: [master, develop, 'feature/**', 'fix/**', 'hotfix/**']
+  pull_request:
+    branches: [master, develop]
+
+jobs:
+  # ═══════════════════════════════════════════════════════════
+  # STAGE 1: GATE
+  # ═══════════════════════════════════════════════════════════
+
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Syntax check
+        run: python -m py_compile keepkeylib/*.py
+
+      - name: Lint summary
+        run: |
+          echo "## 🔑 KeepKey python-keepkey — Lint" >> "$GITHUB_STEP_SUMMARY"
+          echo "" >> "$GITHUB_STEP_SUMMARY"
+          echo "| Check | Status |" >> "$GITHUB_STEP_SUMMARY"
+          echo "|-------|--------|" >> "$GITHUB_STEP_SUMMARY"
+          echo "| Syntax | ✅ PASS |" >> "$GITHUB_STEP_SUMMARY"
+
+  # ═══════════════════════════════════════════════════════════
+  # STAGE 2: TEST — pull published emulator, run pytest
+  # ═══════════════════════════════════════════════════════════
+
+  integration:
+    needs: [lint]
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    services:
+      kkemu:
+        image: kktech/kkemu:latest
+        ports:
+          - 11044:11044/udp
+          - 11045:11045/udp
+          - 5000:5000
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: recursive
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install dependencies
+        run: |
+          pip install --upgrade pip
+          pip install "protobuf>=3.20,<4"
+          pip install -e .
+          pip install pytest semver rlp requests
+
+      - name: Wait for emulator
+        run: |
+          echo "Waiting for emulator bridge on port 5000..."
+          for i in $(seq 1 30); do
+            if curl -sf -X POST http://localhost:5000/exchange/main \
+                 -H 'Content-Type: application/json' \
+                 -d '{"data":""}' > /dev/null 2>&1; then
+              echo "Emulator ready after ${i}s"
+              break
+            fi
+            sleep 1
+          done
+
+      - name: Run integration tests
+        env:
+          KK_TRANSPORT_MAIN: "127.0.0.1:11044"
+          KK_TRANSPORT_DEBUG: "127.0.0.1:11045"
+          PYTHONPATH: "${{ github.workspace }}/keepkeylib:${{ github.workspace }}"
+        run: |
+          cd tests
+          pytest -v --junitxml=junit.xml 2>&1 | tee pytest-output.txt
+          echo "${PIPESTATUS[0]}" > status
+
+      - name: Test summary
+        if: always()
+        run: |
+          XML="tests/junit.xml"
+          echo "## 🔑 KeepKey python-keepkey — Integration Tests" >> "$GITHUB_STEP_SUMMARY"
+          echo "" >> "$GITHUB_STEP_SUMMARY"
+
+          if [ ! -f "$XML" ]; then
+            echo "❌ **No test results found** — suite may have crashed before completion." >> "$GITHUB_STEP_SUMMARY"
+          else
+            TOTAL=$(grep -oP 'tests="\K[0-9]+' "$XML" | head -1)
+            FAILED=$(grep -oP 'failures="\K[0-9]+' "$XML" | head -1)
+            ERRORS=$(grep -oP 'errors="\K[0-9]+' "$XML" | head -1)
+            SKIPPED=$(grep -oP 'skipped="\K[0-9]+' "$XML" | head -1)
+            TIME=$(grep -oP 'time="\K[0-9.]+' "$XML" | head -1)
+
+            TOTAL=${TOTAL:-0}; FAILED=${FAILED:-0}; ERRORS=${ERRORS:-0}; SKIPPED=${SKIPPED:-0}
+            PASSED=$((TOTAL - FAILED - ERRORS - SKIPPED))
+
+            if [ "$FAILED" -eq 0 ] && [ "$ERRORS" -eq 0 ]; then
+              echo "✅ **$PASSED of $TOTAL TESTS PASSED** in ${TIME}s" >> "$GITHUB_STEP_SUMMARY"
+            else
+              echo "❌ **$((FAILED + ERRORS)) of $TOTAL TESTS FAILED**" >> "$GITHUB_STEP_SUMMARY"
+            fi
+
+            echo "" >> "$GITHUB_STEP_SUMMARY"
+            echo "| Metric | Count |" >> "$GITHUB_STEP_SUMMARY"
+            echo "|--------|-------|" >> "$GITHUB_STEP_SUMMARY"
+            echo "| Total | $TOTAL |" >> "$GITHUB_STEP_SUMMARY"
+            echo "| ✅ Passed | $PASSED |" >> "$GITHUB_STEP_SUMMARY"
+            echo "| ⏭️ Skipped | $SKIPPED |" >> "$GITHUB_STEP_SUMMARY"
+            echo "| ❌ Failed | $FAILED |" >> "$GITHUB_STEP_SUMMARY"
+            echo "| 💥 Errors | $ERRORS |" >> "$GITHUB_STEP_SUMMARY"
+
+            # Itemize skipped with reasons
+            python3 -c "import xml.etree.ElementTree as ET,sys;tree=ET.parse(sys.argv[1]);[print(f'| \`{tc.get(\"classname\",\"\")}.{tc.get(\"name\",\"\")}\` | {tc.find(\"skipped\").get(\"message\",tc.find(\"skipped\").text or \"No reason given\")} |') for tc in tree.iter('testcase') if tc.find('skipped') is not None]" "$XML" > /tmp/skip_rows.txt 2>/dev/null || true
+
+            if [ -s /tmp/skip_rows.txt ]; then
+              echo "" >> "$GITHUB_STEP_SUMMARY"
+              echo "### Skipped Tests" >> "$GITHUB_STEP_SUMMARY"
+              echo "| Test | Reason |" >> "$GITHUB_STEP_SUMMARY"
+              echo "|------|--------|" >> "$GITHUB_STEP_SUMMARY"
+              cat /tmp/skip_rows.txt >> "$GITHUB_STEP_SUMMARY"
+            fi
+          fi
+
+          echo "" >> "$GITHUB_STEP_SUMMARY"
+          echo "---" >> "$GITHUB_STEP_SUMMARY"
+          echo "*KeepKey python-keepkey CI*" >> "$GITHUB_STEP_SUMMARY"
+
+      - name: Upload test results
+        uses: mikepenz/action-junit-report@v4
+        if: always()
+        with:
+          report_paths: tests/junit.xml
+          check_name: Integration Tests
+
+      - name: Fail on test failure
+        if: always()
+        run: |
+          STATUS=$(cat tests/status 2>/dev/null || echo "1")
+          [ "$STATUS" = "0" ] || exit 1

build_pb.sh

@@ -3,7 +3,7 @@ CURDIR=$(pwd)
 cd "device-protocol"
 
 echo "Building with protoc version: $(protoc --version)"
-for i in messages messages-ethereum messages-eos messages-nano messages-cosmos messages-ripple messages-binance messages-tendermint messages-thorchain messages-osmosis messages-mayachain types ; do
+for i in messages messages-ethereum messages-eos messages-nano messages-cosmos messages-ripple messages-binance messages-tendermint messages-thorchain messages-osmosis messages-mayachain messages-solana messages-tron messages-ton messages-zcash types ; do
     protoc --python_out=$CURDIR/keepkeylib/ -I/usr/include -I. $i.proto
     i=${i/-/_}
     sed -i -Ee 's/^import ([^.]+_pb2)/from . import \1/' $CURDIR/keepkeylib/"$i"_pb2.py

keepkeylib/client.py

@@ -48,19 +48,27 @@
 from . import messages_solana_pb2 as solana_proto
 from . import messages_tron_pb2 as tron_proto
 from . import messages_ton_pb2 as ton_proto
+from . import messages_zcash_pb2 as zcash_proto
 from . import types_pb2 as types
 from . import eos
 from . import nano
 from .debuglink import DebugLink
 
 
-# try:
-#     from PIL import Image
-#     SCREENSHOT = True
-# except:
-#     SCREENSHOT = False
+import struct as _struct
+import zlib as _zlib
 
-SCREENSHOT = False
+SCREENSHOT = os.environ.get('KEEPKEY_SCREENSHOT', '') == '1'
+
+
+def _write_png(path, width, height, pixels):
+    """Write a minimal grayscale PNG. No Pillow needed."""
+    def _chunk(tag, data):
+        raw = tag + data
+        return _struct.pack('>I', len(data)) + raw + _struct.pack('>I', _zlib.crc32(raw) & 0xffffffff)
+    ihdr = _struct.pack('>IIBBBBB', width, height, 8, 0, 0, 0, 0)
+    raw_data = b''.join(b'\x00' + row for row in pixels)
+    return b'\x89PNG\r\n\x1a\n' + _chunk(b'IHDR', ihdr) + _chunk(b'IDAT', _zlib.compress(raw_data)) + _chunk(b'IEND', b'')
 
 DEFAULT_CURVE = 'secp256k1'
 
@@ -424,17 +432,8 @@ def set_mnemonic(self, mnemonic):
 
     def call_raw(self, msg):
 
-        if SCREENSHOT and self.debug:
-            layout = self.debug.read_layout()
-            im = Image.new("RGB", (128, 64))
-            pix = im.load()
-            for x in range(128):
-                for y in range(64):
-                    rx, ry = 127 - x, 63 - y
-                    if (ord(layout[rx + (ry / 8) * 128]) & (1 << (ry % 8))) > 0:
-                        pix[x, y] = (255, 255, 255)
-            im.save('scr%05d.png' % self.screenshot_id)
-            self.screenshot_id += 1
+        # Screenshot capture disabled in call_raw (captures idle screens, adds latency).
+        # Real confirmation screenshots are captured in callback_ButtonRequest instead.
 
         resp = super(DebugLinkMixin, self).call_raw(msg)
         self._check_request(resp)
@@ -462,6 +461,34 @@ def callback_ButtonRequest(self, msg):
         if self.verbose:
             log("ButtonRequest code: " + get_buttonrequest_value(msg.code))
 
+        # Capture OLED screenshot BEFORE pressing button (confirmation screen)
+        if SCREENSHOT and self.debug:
+            try:
+                layout = self.debug.read_layout()
+                if layout and len(layout) >= 1024:
+                    layout_bytes = len(layout)
+                    height = 64 if layout_bytes >= 2048 else 32
+                    rows = []
+                    for y in range(height):
+                        row = bytearray(256)
+                        for x in range(256):
+                            byte_idx = x + (y // 8) * 256
+                            if byte_idx < layout_bytes:
+                                b = layout[byte_idx] if isinstance(layout[byte_idx], int) else ord(layout[byte_idx])
+                                if (b >> (y % 8)) & 1:
+                                    row[x] = 255
+                        rows.append(bytes(row))
+                    while len(rows) < 64:
+                        rows.append(bytes(256))
+                    screenshot_dir = getattr(self, 'screenshot_dir', os.environ.get('SCREENSHOT_DIR', '.'))
+                    os.makedirs(screenshot_dir, exist_ok=True)
+                    png_path = os.path.join(screenshot_dir, 'btn%05d.png' % self.screenshot_id)
+                    with open(png_path, 'wb') as f:
+                        f.write(_write_png(png_path, 256, 64, rows))
+                    self.screenshot_id += 1
+            except Exception:
+                pass
+
         if self.auto_button:
             if self.verbose:
                 log("Pressing button " + str(self.button))
@@ -1604,6 +1631,112 @@ def ton_sign_tx(self, address_n, raw_tx):
             ton_proto.TonSignTx(address_n=address_n, raw_tx=raw_tx)
         )
 
+    # ── Zcash Orchard ──────────────────────────────────────────
+    @expect(zcash_proto.ZcashOrchardFVK)
+    def zcash_get_orchard_fvk(self, address_n, account=None, show_display=False):
+        kwargs = dict(address_n=address_n, show_display=show_display)
+        if account is not None:
+            kwargs['account'] = account
+        return self.call(zcash_proto.ZcashGetOrchardFVK(**kwargs))
+
+    @session
+    def zcash_sign_pczt(self, address_n, actions, account=None,
+                        total_amount=0, fee=0, branch_id=0x37519621,
+                        header_digest=None, transparent_digest=None,
+                        sapling_digest=None, orchard_digest=None,
+                        orchard_flags=None, orchard_value_balance=None,
+                        orchard_anchor=None, transparent_inputs=None):
+        """Sign a Zcash Orchard shielded transaction via PCZT protocol.
+
+        Phase 2: Sends ZcashSignPCZT, then loops on ZcashPCZTActionAck
+        feeding Orchard actions one at a time.
+        Phase 3: If transparent_inputs provided, handles ZcashTransparentSig
+        loop for transparent-to-shielded (shielding) transactions.
+
+        Args:
+            address_n: ZIP-32 derivation path [32', 133', account']
+            actions: list of dicts, each with keys matching ZcashPCZTAction fields
+            account: account index (default: derived from address_n[2])
+            total_amount: total ZEC in zatoshis (for display)
+            fee: fee in zatoshis (for display)
+            branch_id: consensus branch ID (default NU5)
+            header_digest: 32-byte header digest (enables on-device sighash)
+            transparent_digest: 32-byte transparent digest
+            sapling_digest: 32-byte sapling digest
+            orchard_digest: 32-byte orchard digest
+            orchard_flags: bundle flags byte (enables digest verification)
+            orchard_value_balance: signed i64 value balance
+            orchard_anchor: 32-byte anchor
+
+        Returns:
+            ZcashSignedPCZT with .signatures list and optional .txid
+        """
+        n_actions = len(actions)
+        if n_actions == 0:
+            raise ValueError("Must have at least one action")
+
+        # Build the initial signing request — only send address_n,
+        # let firmware derive account from the path. Only set account
+        # explicitly if the caller passed it.
+        kwargs = dict(
+            address_n=address_n,
+            n_actions=n_actions,
+            total_amount=total_amount,
+            fee=fee,
+            branch_id=branch_id,
+        )
+        if account is not None:
+            kwargs['account'] = account
+        if header_digest is not None:
+            kwargs['header_digest'] = header_digest
+        if transparent_digest is not None:
+            kwargs['transparent_digest'] = transparent_digest
+        if sapling_digest is not None:
+            kwargs['sapling_digest'] = sapling_digest
+        if orchard_digest is not None:
+            kwargs['orchard_digest'] = orchard_digest
+        if orchard_flags is not None:
+            kwargs['orchard_flags'] = orchard_flags
+        if orchard_value_balance is not None:
+            kwargs['orchard_value_balance'] = orchard_value_balance
+        if orchard_anchor is not None:
+            kwargs['orchard_anchor'] = orchard_anchor
+
+        resp = self.call(zcash_proto.ZcashSignPCZT(**kwargs))
+
+        # Phase 2: Orchard action-ack loop — device asks for actions one at a time
+        while isinstance(resp, zcash_proto.ZcashPCZTActionAck):
+            idx = resp.next_index
+            if idx >= n_actions:
+                raise Exception(
+                    "Device requested action index %d but only %d actions provided"
+                    % (idx, n_actions))
+            action = actions[idx]
+            resp = self.call(zcash_proto.ZcashPCZTAction(index=idx, **action))
+
+        # Phase 3: Transparent input signing — device sends back signatures
+        # and may request transparent inputs for shielding transactions
+        transparent_sigs = []
+        while isinstance(resp, zcash_proto.ZcashTransparentSig):
+            transparent_sigs.append(resp)
+            if not transparent_inputs:
+                raise Exception(
+                    "Device sent ZcashTransparentSig but no transparent_inputs provided")
+            if resp.next_index >= len(transparent_inputs):
+                raise Exception(
+                    "Device requested transparent input %d but only %d provided"
+                    % (resp.next_index, len(transparent_inputs)))
+            inp = transparent_inputs[resp.next_index]
+            resp = self.call(zcash_proto.ZcashTransparentInput(**inp))
+
+        if isinstance(resp, proto.Failure):
+            raise Exception("Zcash signing failed: %s" % resp.message)
+
+        if not isinstance(resp, zcash_proto.ZcashSignedPCZT):
+            raise Exception("Unexpected response type: %s" % type(resp))
+
+        return resp
+
 class KeepKeyClient(ProtocolMixin, TextUIMixin, BaseClient):
     pass
 

keepkeylib/debuglink.py

@@ -99,6 +99,19 @@ def read_recovery_auto_completed_word(self):
         obj = self._call(proto.DebugLinkGetState())
         return obj.recovery_auto_completed_word
 
+    def read_recovery_state(self):
+        """Read cipher + auto-completed word + layout in a single call.
+
+        Returns dict with keys: cipher, auto_completed_word, layout
+        Avoids 3 separate DebugLinkGetState round-trips per character.
+        """
+        obj = self._call(proto.DebugLinkGetState())
+        return {
+            'cipher': obj.recovery_cipher,
+            'auto_completed_word': obj.recovery_auto_completed_word,
+            'layout': obj.layout,
+        }
+
     def read_memory_hashes(self):
         obj = self._call(proto.DebugLinkGetState())
         return (obj.firmware_hash, obj.storage_hash)