Download & decode the entire iPixel Color (Heaton) LED-matrix animation library — zero dependencies, pure-Python AES bundled.

ipixel_grab.py

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT

"""
ipixel_grab.py - mirror the iPixel Color ("Heaton") cloud material library.

Downloads and decodes every animation/picture the iPixel Color app can show
on an LED matrix, straight from Heaton's cloud. Each category is written to 
its own folder, and runs are resumable.

Background & how the format was reverse-engineered:
  https://github.com/lucagoc/pypixelcolor/discussions/60

Zero third-party dependencies: pure-Python AES is bundled, so this runs on a
stock Python 3 install (no `pip install` required).

For a given panel size it will:
  1. fetch the category manifest (sucai_define.json)
  2. for every category (animations + pictures)
       - page through the cloud API to list every file
       - create a directory
       - download + decode each file into it
  3. repeat for the next category

Protocol recovered from the app (CryptographicParsingTool / AESUtils):
  * AES-256-CBC, key == sign app_key == "5uMoiJzoUCtI7KWl4U5yWWcFK6OEb6Nz",
    fixed IV b"0000000000000000", PKCS7, standard base64.
  * request body  = AES( sorted "k=v&..." url-ish )         (base64)
  * sign (query)  = md5( urlish( sorted(params+random+timestamp+app_key) ) )
  * response body = AES( JSON )                              (base64)
  * download files: strip 32-char token each end -> url-decode
                    -> reverse the string -> base64-decode -> raw image

Usage:
  python3 ipixel_grab.py                       # 64x20, everything, ./ipixel_library
  python3 ipixel_grab.py --size 96x16 -o out.  # 96x16, everything, ./out
  python3 ipixel_grab.py --only eyes,emoji     # subset (languageKey or requestKey)
  python3 ipixel_grab.py --types ani           # animations only
  python3 ipixel_grab.py --limit 3             # first 3 per category (a dry-ish run)
  python3 ipixel_grab.py --secure             # re-enable TLS cert verification
  python3 ipixel_grab.py --help               # full option list

TLS note: certificate verification is OFF by default, because the stock macOS
python.org build often ships without a usable CA store ("unable to get local
issuer certificate"). Pass --secure once your CA store works (e.g. after
running Install Certificates.command or `pip install certifi`).
"""

import argparse
import base64
import hashlib
import json
import os
import random
import string
import sys
import time
import ssl
import urllib.parse
import urllib.request

# ==========================================================================
# bundled pure-Python AES-256-CBC (verified against FIPS-197 test vectors)
# ==========================================================================
def _gmul(a, b):
    p = 0
    for _ in range(8):
        if b & 1:
            p ^= a
        hi = a & 0x80
        a = (a << 1) & 0xFF
        if hi:
            a ^= 0x1B
        b >>= 1
    return p

_INV_GF = [0] * 256
for _a in range(1, 256):
    for _b in range(1, 256):
        if _gmul(_a, _b) == 1:
            _INV_GF[_a] = _b
            break

def _affine(x):
    s = x
    for i in range(1, 5):
        x ^= ((s << i) | (s >> (8 - i))) & 0xFF
    return x ^ 0x63

SBOX = [_affine(_INV_GF[i]) for i in range(256)]
INV_SBOX = [0] * 256
for _i, _v in enumerate(SBOX):
    INV_SBOX[_v] = _i

def _expand(key):
    Nk = len(key) // 4
    Nr = {4: 10, 6: 12, 8: 14}[Nk]
    w = [list(key[4 * i:4 * i + 4]) for i in range(Nk)]
    rc = 1
    for i in range(Nk, 4 * (Nr + 1)):
        t = list(w[-1])
        if i % Nk == 0:
            t = t[1:] + t[:1]
            t = [SBOX[b] for b in t]
            t[0] ^= rc
            rc = _gmul(rc, 2)
        elif Nk > 6 and i % Nk == 4:
            t = [SBOX[b] for b in t]
        w.append([w[i - Nk][j] ^ t[j] for j in range(4)])
    return w, Nr

def _rk(w, r):
    o = [0] * 16
    for c in range(4):
        for row in range(4):
            o[row + 4 * c] = w[4 * r + c][row]
    return o

def _enc_block(block, w, Nr):
    s = list(block)
    k = _rk(w, 0)
    for i in range(16):
        s[i] ^= k[i]
    for rnd in range(1, Nr):
        s = [SBOX[x] for x in s]
        s = [s[(i % 4) + 4 * ((i // 4 + i % 4) % 4)] for i in range(16)]
        ns = list(s)
        for c in range(4):
            a = s[4 * c:4 * c + 4]
            ns[4 * c + 0] = _gmul(a[0], 2) ^ _gmul(a[1], 3) ^ a[2] ^ a[3]
            ns[4 * c + 1] = a[0] ^ _gmul(a[1], 2) ^ _gmul(a[2], 3) ^ a[3]
            ns[4 * c + 2] = a[0] ^ a[1] ^ _gmul(a[2], 2) ^ _gmul(a[3], 3)
            ns[4 * c + 3] = _gmul(a[0], 3) ^ a[1] ^ a[2] ^ _gmul(a[3], 2)
        s = ns
        k = _rk(w, rnd)
        for i in range(16):
            s[i] ^= k[i]
    s = [SBOX[x] for x in s]
    s = [s[(i % 4) + 4 * ((i // 4 + i % 4) % 4)] for i in range(16)]
    k = _rk(w, Nr)
    for i in range(16):
        s[i] ^= k[i]
    return bytes(s)

def _dec_block(block, w, Nr):
    s = list(block)
    k = _rk(w, Nr)
    for i in range(16):
        s[i] ^= k[i]
    for rnd in range(Nr - 1, 0, -1):
        s = [s[(i % 4) + 4 * ((i // 4 - i % 4) % 4)] for i in range(16)]
        s = [INV_SBOX[x] for x in s]
        k = _rk(w, rnd)
        for i in range(16):
            s[i] ^= k[i]
        ns = list(s)
        for c in range(4):
            a = s[4 * c:4 * c + 4]
            ns[4 * c + 0] = _gmul(a[0], 14) ^ _gmul(a[1], 11) ^ _gmul(a[2], 13) ^ _gmul(a[3], 9)
            ns[4 * c + 1] = _gmul(a[0], 9) ^ _gmul(a[1], 14) ^ _gmul(a[2], 11) ^ _gmul(a[3], 13)
            ns[4 * c + 2] = _gmul(a[0], 13) ^ _gmul(a[1], 9) ^ _gmul(a[2], 14) ^ _gmul(a[3], 11)
            ns[4 * c + 3] = _gmul(a[0], 11) ^ _gmul(a[1], 13) ^ _gmul(a[2], 9) ^ _gmul(a[3], 14)
        s = ns
    s = [s[(i % 4) + 4 * ((i // 4 - i % 4) % 4)] for i in range(16)]
    s = [INV_SBOX[x] for x in s]
    k = _rk(w, 0)
    for i in range(16):
        s[i] ^= k[i]
    return bytes(s)

def aes_cbc_encrypt(data, key, iv):
    w, Nr = _expand(key)
    padlen = 16 - (len(data) % 16)
    data = data + bytes([padlen]) * padlen        # PKCS7
    out, prev = bytearray(), iv
    for i in range(0, len(data), 16):
        x = bytes(a ^ b for a, b in zip(data[i:i + 16], prev))
        prev = _enc_block(x, w, Nr)
        out += prev
    return bytes(out)

def aes_cbc_decrypt(data, key, iv):
    w, Nr = _expand(key)
    out, prev = bytearray(), iv
    for i in range(0, len(data), 16):
        blk = data[i:i + 16]
        p = _dec_block(blk, w, Nr)
        out += bytes(a ^ b for a, b in zip(p, prev))
        prev = blk
    return bytes(out[:-out[-1]])                    # strip PKCS7


# ==========================================================================
# constants (from the decompiled app)
# ==========================================================================
KEY = b"5uMoiJzoUCtI7KWl4U5yWWcFK6OEb6Nz"
IV = b"0000000000000000"
APP_KEY = "5uMoiJzoUCtI7KWl4U5yWWcFK6OEb6Nz"
API = "https://manage.heaton.com.cn/api/rm/getMaterialUnderCategory"
MANIFEST = "http://app.heaton.cn/sucai_define.json"
UA = "WIFI_LED/2 CFNetwork/3860.600.12 Darwin/25.5.0"
TOKEN = 32  # opaque chars stripped from each end of a downloaded file

# TLS context: verification is OFF by default (--secure turns it on).
# When secure, honour certifi if present, else the system default store.
_CTX = None

def _make_ctx(insecure: bool):
    if insecure:
        c = ssl.create_default_context()
        c.check_hostname = False
        c.verify_mode = ssl.CERT_NONE
        return c
    try:
        import certifi
        return ssl.create_default_context(cafile=certifi.where())
    except Exception:
        return ssl.create_default_context()


# ==========================================================================
# crypto / signing wrappers
# ==========================================================================
def _urlish(s: str) -> str:
    out = urllib.parse.quote_plus(s, safe="*-._")
    return out.replace("%26", "&").replace("%3D", "=").replace("%3F", "?")


def _encrypt(params: dict) -> str:
    joined = "&".join(f"{k}={params[k]}" for k in sorted(params))
    return base64.b64encode(aes_cbc_encrypt(_urlish(joined).encode(), KEY, IV)).decode()


def _decrypt(b64: str) -> str:
    return aes_cbc_decrypt(base64.b64decode(b64), KEY, IV).decode("utf-8")


def _sign(params: dict, rnd: str, ts: str) -> str:
    m = dict(params, random=rnd, timestamp=ts, app_key=APP_KEY)
    joined = "&".join(f"{k}={m[k]}" for k in sorted(m))
    return hashlib.md5(_urlish(joined).encode()).hexdigest().lower()


def _rand(n: int = 8) -> str:
    return "".join(random.choice(string.ascii_letters + string.digits) for _ in range(n))


def _http(url, data=None, method="GET", timeout=30, headers=None) -> bytes:
    req = urllib.request.Request(url, data=data, method=method,
                                 headers=headers or {"User-Agent": UA})
    ctx = _CTX if url.lower().startswith('https') else None
    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
        raw = r.read()
        if (r.headers.get("Content-Encoding") or "").lower() == "gzip":
            import gzip
            raw = gzip.decompress(raw)
    return raw


# ==========================================================================
# api
# ==========================================================================
def api_call(params: dict, timeout=30) -> dict:
    ts, rnd = str(int(time.time())), _rand()
    body = _encrypt(params).encode()
    qs = urllib.parse.urlencode({"sign": _sign(params, rnd, ts),
                                 "timestamp": ts, "random": rnd})
    raw = _http(f"{API}?{qs}", data=body, method="POST",
                headers={"Content-Type": "text/plain", "Accept": "*/*",
                         "User-Agent": UA}, timeout=timeout)
    return json.loads(_decrypt(raw.decode("utf-8", "replace")))


def list_category(category_name, size_wh, label, type_, count=60):
    w, h = size_wh
    page, seen, total = 1, 0, None
    while True:
        params = {"appid": "137", "sort": "1", "page": str(page),
                  "count": str(count), "category_name": category_name,
                  "type": type_, "label": label,
                  "width": str(w), "height": str(h),
                  "file_lang": "none,en", "filter_tags": ""}
        data = api_call(params).get("data", {}) or {}
        recs = data.get("records", []) or []
        total = data.get("totalCount", total)
        if not recs:
            break
        for r in recs:
            yield r
        seen += len(recs)
        if total is not None and seen >= total:
            break
        page += 1


# ==========================================================================
# file download + decode  (matches app's getDecryptedFile exactly)
# ==========================================================================
def download_decode(url: str, timeout=30) -> bytes:
    text = _http(url, timeout=timeout).decode("latin-1")
    if len(text) <= 64:
        raise ValueError("payload too short")
    core = text[TOKEN:len(text) - TOKEN].replace("+", " ")
    core = urllib.parse.unquote(core)
    core = core[::-1].strip().replace("\r", "").replace("\n", "")
    return base64.b64decode(core)


# ==========================================================================
# manifest / categories
# ==========================================================================
def load_manifest(size, url=None, path=None):
    if path:
        data = json.load(open(path, encoding="utf-8"))
    else:
        data = json.loads(_http(url or MANIFEST).decode("utf-8"))
    for entry in data:
        if entry.get("size") == size:
            return entry
    sizes = ", ".join(sorted({e.get("size", "?") for e in data}))
    raise SystemExit(f"size {size!r} not in manifest. available: {sizes}")


def categories_for(entry, types):
    type_param = {"ani": "动画", "pic": "图片",
                  "aniVertical": "动画", "picVertical": "图片"}
    out = []
    for group in ("ani", "pic", "aniVertical", "picVertical"):
        if group.replace("Vertical", "") not in types:
            continue
        for item in entry.get("categorys", {}).get(group, []) or []:
            out.append((group, type_param[group],
                        item.get("languageKey") or item["requestKey"],
                        item["requestKey"]))
    return out


def _safe(name: str) -> str:
    return "".join(c if c not in '/\\:*?"<>|' else "_" for c in name).strip() or "x"


# ==========================================================================
# main
# ==========================================================================
def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--size", default="64x20", help="panel size (default 64x20)")
    ap.add_argument("-o", "--out", default="ipixel_library", help="output root dir")
    ap.add_argument("--types", default="ani,pic", help="comma list of ani,pic")
    ap.add_argument("--only", default="",
                    help="comma list of category langKeys/requestKeys to include")
    ap.add_argument("--count", type=int, default=60, help="API page size")
    ap.add_argument("--limit", type=int, default=0,
                    help="max files per category (0 = all; use for a test run)")
    ap.add_argument("--delay", type=float, default=0.1,
                    help="seconds between downloads (be polite)")
    ap.add_argument("--retries", type=int, default=3, help="download retries")
    ap.add_argument("--overwrite", action="store_true",
                    help="re-download files that already exist")
    ap.add_argument("--secure", action="store_true",
                    help="enable TLS certificate verification "
                         "(OFF by default; turn on if you have a working CA store)")
    ap.add_argument("--manifest-url", default=MANIFEST)
    ap.add_argument("--manifest-file", default=None,
                    help="use a local sucai_define.json instead of fetching")
    args = ap.parse_args()

    global _CTX
    _CTX = _make_ctx(insecure=not args.secure)

    types = {t.strip() for t in args.types.split(",") if t.strip()}
    only = {s.strip() for s in args.only.split(",") if s.strip()}

    entry = load_manifest(args.size, args.manifest_url, args.manifest_file)
    label = entry.get("label", "ALL,YK")
    w, h = (int(x) for x in args.size.lower().split("x"))
    cats = categories_for(entry, types)
    if only:
        cats = [c for c in cats if c[2] in only or c[3] in only]
    if not cats:
        raise SystemExit("no categories matched")

    root = os.path.abspath(args.out)
    os.makedirs(root, exist_ok=True)
    print(f"# size {args.size}  label {label}  ->  {root}")
    print(f"# {len(cats)} categories: " +
          ", ".join(f"{g}/{lk}" for g, _, lk, _ in cats))

    grand_ok = grand_fail = 0
    summary = []
    for group, type_param, lang, req in cats:
        top = "animations" if type_param == "动画" else "pictures"
        cdir = os.path.join(root, top, _safe(lang))
        os.makedirs(cdir, exist_ok=True)
        print(f"\n== {top}/{lang}  ({req})")

        try:
            records = list(list_category(req, (w, h), label, type_param, args.count))
        except Exception as exc:
            print(f"   ! listing failed: {exc}")
            summary.append((f"{top}/{lang}", 0, 0, "list-failed"))
            continue
        if args.limit:
            records = records[:args.limit]
        print(f"   {len(records)} files")

        with open(os.path.join(cdir, "_index.tsv"), "w", encoding="utf-8") as index:
            index.write("file_id\tlabel\tformat\turl\n")
            ok = fail = skip = 0
            for i, rec in enumerate(records, 1):
                fid = rec.get("file_id", i)
                fmt = (rec.get("format") or "gif").lstrip(".")
                url = rec["file_path"]
                index.write(f'{fid}\t{rec.get("label","")}\t{fmt}\t{url}\n')
                dest = os.path.join(cdir, f"{fid}.{fmt}")
                if os.path.exists(dest) and not args.overwrite:
                    skip += 1
                    continue
                for attempt in range(1, args.retries + 1):
                    try:
                        with open(dest, "wb") as f:
                            f.write(download_decode(url))
                        ok += 1
                        break
                    except Exception as exc:
                        if attempt == args.retries:
                            print(f"   ! {fid}: {exc}")
                            fail += 1
                        else:
                            time.sleep(0.5 * attempt)
                if args.delay:
                    time.sleep(args.delay)
                if i % 25 == 0:
                    print(f"   ... {i}/{len(records)}")

        print(f"   done: {ok} downloaded, {skip} skipped, {fail} failed")
        grand_ok += ok
        grand_fail += fail
        summary.append((f"{top}/{lang}", len(records), ok + skip, fail))

    print("\n==== summary ====")
    for name, total, have, fail in summary:
        print(f"  {name:24} {have}/{total} present" +
              (f"  ({fail} failed)" if fail else ""))
    print(f"  total downloaded this run: {grand_ok}, failures: {grand_fail}")
    print(f"  library at: {root}")
    return 1 if grand_fail else 0


if __name__ == "__main__":
    sys.exit(main())