feat: add verified 6-bit header codec

cq2_sdk/src/cq2_sdk/transport/encode6.py

@@ -0,0 +1,69 @@
+"""Verified six-bit encoding used for the CQ2 TCP wire header.
+
+The table is the first 256 bytes of the archived client extraction
+``Encode6BitBuf_替换表.bin`` (SHA-256:
+bd8e0d6b48cff81a2e25e497a86e5e5ef33fedab7cb4af6a1e0db05ddd59fe1e).
+It is embedded so the runtime never depends on reverse-engineering artifacts.
+"""
+from __future__ import annotations
+
+
+ENC_TABLE = bytes.fromhex(
+    "8c870d85d46463e5ba7eb8689d9ff5bca0e33a2219213978ee27361574c797c9"
+    "cee27b4c98a1c25941c01e2e95ebde691d5b53daf40a4fbbb724330fc8842989"
+    "3c1c0849c6fecc233ee14e8b13e71a5dcfb1478fd8724b936e734d94dd8214a7"
+    "03f9f1c58d792ac4dc605fd762b5e9b3b612a832d9c36a754aa20c26915aad6d"
+    "4410b4461b668120fd7f88259c71d3e680e4fa429b3701fcdb456bfb56f0af9a"
+    "bfabd6cd02f27caab292ff572f86a67d351734d50e65090528ca48318e2ddf52"
+    "f61fa450764018048a162bae433fd0cb6c5554969930675e2cace07ae85890be"
+    "a56fb070ec615c063b77c107eaa9f811bdf300ed83ef3da3519e38f70bb9d2d1"
+)
+
+_inverse = bytearray(256)
+for _plain, _encoded in enumerate(ENC_TABLE):
+    _inverse[_encoded] = _plain
+INV_TABLE = bytes(_inverse)
+
+
+def encode6(src: bytes) -> bytes:
+    """Encode bytes into the client's printable 6-bit alphabet."""
+    dst = bytearray()
+    accumulator = 0
+    pending_bits = 0
+    for byte in src:
+        substituted = ENC_TABLE[byte] ^ 9
+        next_pending_bits = pending_bits + 2
+        accumulator |= substituted >> next_pending_bits
+        dst.append((accumulator & 0x3F) + 0x3C)
+        accumulator = (substituted << ((6 - pending_bits) & 0xFF)) >> 2 & 0x3F
+        if pending_bits >= 4:
+            dst.append(accumulator + 0x3C)
+            accumulator = 0
+            pending_bits = 0
+        else:
+            pending_bits = next_pending_bits
+    if pending_bits:
+        dst.append(accumulator + 0x3C)
+    return bytes(dst)
+
+
+def decode6(data: bytes, *, expected_size: int | None = None) -> bytes:
+    """Decode a printable 6-bit client header, validating its wire alphabet."""
+    bits = 0
+    bit_count = 0
+    substituted = bytearray()
+    for character in data:
+        value = character - 0x3C
+        if not 0 <= value <= 0x3F:
+            raise ValueError(f"byte outside 6-bit wire range: 0x{character:02x}")
+        bits = (bits << 6) | value
+        bit_count += 6
+        while bit_count >= 8:
+            bit_count -= 8
+            substituted.append((bits >> bit_count) & 0xFF)
+    decoded = bytes(INV_TABLE[byte ^ 9] for byte in substituted)
+    if expected_size is not None and len(decoded) != expected_size:
+        raise ValueError(
+            f"decoded header size mismatch: expected {expected_size}, got {len(decoded)}"
+        )
+    return decoded

cq2_sdk/tests/unit/test_framing.py

@@ -1,8 +1,31 @@
 import pytest
 from cq2_sdk.transport.framing import encode_frame, decode_frame, FrameHeader, HEADER_SIZE
+from cq2_sdk.transport.encode6 import decode6, encode6
 from cq2_sdk.messages import constants as C
 
 
+HEARTBEAT_HEADER = bytes.fromhex(
+    "0000000060ea0000b0cdac6a000000000000000095000000000000000000000070000000"
+)
+HEARTBEAT_WIRE_HEADER = bytes.fromhex(
+    "5d5452415d4c67445d545078495e55525d5452415d545241"
+    "5d54507a5d5452415d5452415d5452415d54527b5d545241"
+)
+
+
+def test_encode6_matches_captured_heartbeat_header():
+    assert encode6(HEARTBEAT_HEADER) == HEARTBEAT_WIRE_HEADER
+
+
+def test_decode6_restores_captured_heartbeat_header():
+    assert decode6(HEARTBEAT_WIRE_HEADER, expected_size=36) == HEARTBEAT_HEADER
+
+
+def test_decode6_rejects_non_wire_character():
+    with pytest.raises(ValueError, match="outside 6-bit wire range"):
+        decode6(b"<~")
+
+
 def test_header_is_24_bytes():
     raw = encode_frame(1, 0, 0, 0, 0, b"")
     assert len(raw) == 24