Composite PQC: make sign/decrypt usable, conform to draft-10, and stop reporting false success

onlykey/age_plugin/bech32.py

@@ -0,0 +1,89 @@
+"""Bech32 encoding for age recipients/identities.
+
+Simplified implementation for age1onlykey1... / AGE-PLUGIN-ONLYKEY-1...
+format - deliberately has no length cap (unlike the standard `bech32` PyPI
+package, which enforces BIP-173's 90-character limit), since a 1216-byte
+X-Wing recipient encodes to something far longer than that.
+
+Extracted from cli.py (where this originated, used correctly there for the
+slot-based recipient/identity encoding) into its own module so
+derived_xwing.py can use the same encoder for derived identities without a
+circular import between the two.
+"""
+
+BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
+
+
+def _bech32_polymod(values):
+    """Internal function for Bech32 checksum."""
+    GEN = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
+    chk = 1
+    for v in values:
+        b = chk >> 25
+        chk = ((chk & 0x1FFFFFF) << 5) ^ v
+        for i in range(5):
+            chk ^= GEN[i] if ((b >> i) & 1) else 0
+    return chk
+
+
+def _bech32_hrp_expand(hrp):
+    return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp]
+
+
+def _bech32_create_checksum(hrp, data):
+    values = _bech32_hrp_expand(hrp) + data
+    polymod = _bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ 1
+    return [(polymod >> 5 * (5 - i)) & 31 for i in range(6)]
+
+
+def _bech32_verify_checksum(hrp, data):
+    return _bech32_polymod(_bech32_hrp_expand(hrp) + data) == 1
+
+
+def _convertbits(data, frombits, tobits, pad=True):
+    """General power-of-2 base conversion."""
+    acc = 0
+    bits = 0
+    ret = []
+    maxv = (1 << tobits) - 1
+    for value in data:
+        if value < 0 or (value >> frombits):
+            return None
+        acc = (acc << frombits) | value
+        bits += frombits
+        while bits >= tobits:
+            bits -= tobits
+            ret.append((acc >> bits) & maxv)
+    if pad:
+        if bits:
+            ret.append((acc << (tobits - bits)) & maxv)
+    elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
+        return None
+    return ret
+
+
+def bech32_encode(hrp: str, data: bytes) -> str:
+    """Encode bytes as Bech32."""
+    values = _convertbits(list(data), 8, 5)
+    checksum = _bech32_create_checksum(hrp, values)
+    return hrp + "1" + "".join(BECH32_CHARSET[d] for d in values + checksum)
+
+
+def bech32_decode(bech: str):
+    """Decode Bech32 string to (hrp, data_bytes)."""
+    if any(ord(x) < 33 or ord(x) > 126 for x in bech):
+        return None, None
+    bech = bech.lower()
+    pos = bech.rfind("1")
+    if pos < 1 or pos + 7 > len(bech):
+        return None, None
+    hrp = bech[:pos]
+    data = [BECH32_CHARSET.find(x) for x in bech[pos + 1:]]
+    if -1 in data:
+        return None, None
+    if not _bech32_verify_checksum(hrp, data):
+        return None, None
+    decoded = _convertbits(data[:-6], 5, 8, False)
+    if decoded is None:
+        return None, None
+    return hrp, bytes(decoded)

onlykey/age_plugin/cli.py

@@ -23,93 +23,13 @@
 from onlykey.age_plugin import (
     __version__, PLUGIN_NAME, DEFAULT_XWING_SLOT, validate_ecc_slot,
 )
+from onlykey.age_plugin.bech32 import bech32_encode, bech32_decode
 from onlykey.age_plugin.protocol import (
     Stanza, b64encode_no_pad, b64decode_no_pad,
     run_identity_v1, run_recipient_v1,
 )
 
 
-# Bech32 encoding for age recipients/identities
-# Simplified implementation for age1onlykey1... format
-
-BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
-
-
-def _bech32_polymod(values):
-    """Internal function for Bech32 checksum."""
-    GEN = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
-    chk = 1
-    for v in values:
-        b = chk >> 25
-        chk = ((chk & 0x1FFFFFF) << 5) ^ v
-        for i in range(5):
-            chk ^= GEN[i] if ((b >> i) & 1) else 0
-    return chk
-
-
-def _bech32_hrp_expand(hrp):
-    return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp]
-
-
-def _bech32_create_checksum(hrp, data):
-    values = _bech32_hrp_expand(hrp) + data
-    polymod = _bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ 1
-    return [(polymod >> 5 * (5 - i)) & 31 for i in range(6)]
-
-
-def _bech32_verify_checksum(hrp, data):
-    return _bech32_polymod(_bech32_hrp_expand(hrp) + data) == 1
-
-
-def _convertbits(data, frombits, tobits, pad=True):
-    """General power-of-2 base conversion."""
-    acc = 0
-    bits = 0
-    ret = []
-    maxv = (1 << tobits) - 1
-    for value in data:
-        if value < 0 or (value >> frombits):
-            return None
-        acc = (acc << frombits) | value
-        bits += frombits
-        while bits >= tobits:
-            bits -= tobits
-            ret.append((acc >> bits) & maxv)
-    if pad:
-        if bits:
-            ret.append((acc << (tobits - bits)) & maxv)
-    elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
-        return None
-    return ret
-
-
-def bech32_encode(hrp: str, data: bytes) -> str:
-    """Encode bytes as Bech32."""
-    values = _convertbits(list(data), 8, 5)
-    checksum = _bech32_create_checksum(hrp, values)
-    return hrp + "1" + "".join(BECH32_CHARSET[d] for d in values + checksum)
-
-
-def bech32_decode(bech: str):
-    """Decode Bech32 string to (hrp, data_bytes)."""
-    if any(ord(x) < 33 or ord(x) > 126 for x in bech):
-        return None, None
-    bech = bech.lower()
-    pos = bech.rfind("1")
-    if pos < 1 or pos + 7 > len(bech):
-        return None, None
-    hrp = bech[:pos]
-    data = [BECH32_CHARSET.find(x) for x in bech[pos + 1 :]]
-    if -1 in data:
-        return None, None
-    if not _bech32_verify_checksum(hrp, data):
-        return None, None
-    decoded = _convertbits(data[:-6], 5, 8, False)
-    if decoded is None:
-        return None, None
-    return hrp, bytes(decoded)
-
-
 # HRP for OnlyKey age recipients and identities
 RECIPIENT_HRP = "age1onlykey"
 IDENTITY_HRP = "age-plugin-onlykey-"  # uppercase AGE-PLUGIN-ONLYKEY- in file

onlykey/age_plugin/derived_xwing.py

@@ -15,7 +15,6 @@
   DERIVE_SHAREDSEC  -> [ ss_X(32) | mlkem_seed(32) ]
 """
 
-import base64
 import hashlib
 
 from kyber_py.ml_kem import ML_KEM_768
@@ -81,24 +80,50 @@ def ct_x_of(ciphertext):
 # Distinguishes a derived identity from a slot identity so age-plugin-onlykey
 # can support BOTH models (like SSH/GPG). A derived identity carries the label;
 # the key is reproduced on demand from (OnlyKey web-derivation key, label, RPID).
-_DERIVED_PREFIX = "AGE-PLUGIN-ONLYKEY-DERIVED-"
+#
+# Real bech32 (cli.py's bech32_encode/decode, extracted to bech32.py so both
+# modules can share it without a circular import), matching the slot-based
+# encode_identity()'s scheme - NOT the naive base32-with-no-checksum
+# concatenation this used to be. That produced strings like
+# "AGE-PLUGIN-ONLYKEY-DERIVED-<base32>" with no "1" bech32 separator and no
+# checksum, which `age` itself rejects outright before ever handing off to
+# the plugin ("invalid identity encoding: separator '1' at invalid
+# position") - observed running an actual `age -d -i <file>` against one.
+#
+# Second, deeper issue found the same way, fixed here too: the HRP can't be
+# a distinct "age-plugin-onlykey-derived-" string either, even bech32-valid.
+# `age` picks which plugin *binary* to run from the "AGE-PLUGIN-<NAME>-"
+# prefix text itself (name -> `age-plugin-<name>`), so a
+# "AGE-PLUGIN-ONLYKEY-DERIVED-1..." identity made `age` look for a
+# nonexistent `age-plugin-onlykey-derived` executable instead of invoking
+# the real, installed `age-plugin-onlykey` - confirmed live
+# ("couldn't start plugin: exec: ... not found in $PATH"). The HRP has to
+# be *exactly* cli.py's IDENTITY_HRP (kept as a literal here, not imported,
+# to avoid a cross-module dependency for one constant - the two must match,
+# noted in both places). Slot vs. derived identities are instead
+# distinguished by a marker byte in the decoded payload: cli.py's
+# decode_identity() only ever produces `data[0]` in {a valid slot 1-132} or
+# {IDENTITY_VERSION==1}, so 0xFF as data[0] is unambiguous and safe - the
+# slot decoder raises ValueError on it either way (wrong length or
+# unrecognized version), which callers already catch and skip.
+from onlykey.age_plugin.bech32 import bech32_encode, bech32_decode
+
+_IDENTITY_HRP = "age-plugin-onlykey-"  # MUST match cli.py's IDENTITY_HRP
+_DERIVED_MARKER = 0xFF
 
 
 def encode_identity(label):
     """Encode a derived identity string for a label (used with `age -i`)."""
     if not isinstance(label, str) or not label:
         raise ValueError("derived identity needs a non-empty label")
-    b32 = base64.b32encode(label.encode("utf-8")).decode("ascii").rstrip("=")
-    return _DERIVED_PREFIX + b32.upper()
+    payload = bytes([_DERIVED_MARKER]) + label.encode("utf-8")
+    return bech32_encode(_IDENTITY_HRP, payload).upper()
 
 
 def decode_identity(s):
     """Decode a derived identity string -> {'derived': True, 'label': str},
     or None if `s` is not a derived identity (caller falls back to slot decode)."""
-    s = str(s).strip().upper()
-    if not s.startswith(_DERIVED_PREFIX):
+    hrp, data = bech32_decode(str(s).strip().lower())
+    if hrp != _IDENTITY_HRP or not data or data[0] != _DERIVED_MARKER:
         return None
-    b32 = s[len(_DERIVED_PREFIX):]
-    b32 += "=" * (-len(b32) % 8)
-    label = base64.b32decode(b32).decode("utf-8")
-    return {"derived": True, "label": label}
+    return {"derived": True, "label": data[1:].decode("utf-8")}

onlykey/age_plugin/onlykey_hid.py

@@ -96,17 +96,25 @@ def _read_response(self, expected_size=0, timeout_ms=10000):
         while time.time() < deadline:
             try:
                 data = self.ok.read_bytes(64, timeout_ms=2000)
-                if data:
-                    text = bytes(data).decode("ascii", errors="ignore")
-                    if text.startswith("Error"):
-                        raise RuntimeError(f"OnlyKey: {text.strip()}")
-                    result.extend(data)
-                    if expected_size and len(result) >= expected_size:
-                        break
             except Exception:
-                if result:
-                    break
+                # A single read timing out mid-stream doesn't mean the
+                # device is done sending - keep polling until the real
+                # deadline. Bailing out early here (as soon as `result` was
+                # non-empty) was truncating multi-packet responses like the
+                # 1216-byte X-Wing pubkey whenever one 2s read happened to
+                # time out before the next packet arrived. This only guards
+                # the read() call itself - a real device-reported error
+                # (below) still needs to propagate immediately, not get
+                # silently swallowed by a broad except around both.
                 continue
+            if not data:
+                continue
+            text = bytes(data).decode("ascii", errors="ignore")
+            if text.startswith("Error"):
+                raise RuntimeError(f"OnlyKey: {text.strip()}")
+            result.extend(data)
+            if expected_size and len(result) >= expected_size:
+                break
 
         return bytes(result[:expected_size] if expected_size else result)