TCP OTA 客户端接入手册

┌──────────┐   FRP 公网隧道    ┌──────────┐   TCP 长连接    ┌──────────────┐
│  设备端   │ ◄──────────────► │ tcp_mux  │ ◄────────────► │ OTA 服务      │
│ (IoT/App)│ tcp.qftek.top    │  (转发)   │ 127.0.0.1:8011 │ (业务)        │
│          │  :32200          │  :8000    │                │               │
└──────────┘                  └──────────┘                └──────────────┘

┌──────────┬────────┬────────┬────────┬──────────┬─────────┐
│ 帧同步字  │ 帧类型  │ 服务类型│ 会话ID  │ 数据长度  │ 数据负载 │
│  6 bytes │ 1 byte │ 1 byte │ 2 bytes│ 4 bytes  │ N bytes │
└──────────┴────────┴────────┴────────┴──────────┴─────────┘
 offset 0   off 6    off 7    off 8    off 10     off 14

设备 ──► SESS_REQ   frame_type=0x01, svc_type=1, sid=0, 负载空
设备 ◄── SESS_RESP  frame_type=0x02, sid=<分配值>   ← 读帧头第 8~9 字节

{"ok":0,"err":3,"msg":"unknown product 'Foo'"}

{"ok":1,"pver":1,"chunk":8192,"min":128,"max":65535}

{"ok":1,"latest":"1.1.5","upd":1,"busy":0,"force":0,"hw":"1.0.0",
 "size":36875511,"md5":"82c6aa016c033b037492bd3cae32df79",
 "delta":1,"dsize":2311859}

{"ok":1,"sid":12,"trans":"D","size":2311859,"total":142,
 "chunk":8192,"md5":"82c6aa016c033b037492bd3cae32df79"}

{"ok":1,"cnt":16,"next":131072,"eof":0,"done":131072,"size":36875511,"bin":1,
 "chunks":[{"off":0,"len":8192,"crc":"a1b2c3d4"},
           {"off":8192,"len":8192,"crc":"e5f6a7b8"}]}

{"ok":1,"cnt":1,"next":8192,"eof":0,"done":8192,
 "chunks":[{"off":0,"len":8192,"d":"48656c6c6f..."}]}

{"ok":1,"cnt":0,"again":1,"next":<原偏移>,"eof":0}

{"ok":1,"sid":12,"size":36875511,"md5":"82c6aa016c033b037492bd3cae32df79"}

请求：{"act":"abort","sid":12}
响应：{"ok":1}  或  {"ok":0,"err":N,"msg":"..."}

请求：{"act":"report","event":"patch_md5_mismatch","msg":"..."}
响应：{"ok":1}  或  {"ok":0,"err":N,"msg":"..."}

{"ok":1,"content":"v1.1.5\n\n### 新增\n- ...\n"}

握手：SESS_REQ ──► ◄── SESS_RESP (sid=N)

hello    {"act":"hello","chunk":8192}                    （可选，协商分片）
check    {"act":"check","product":P,"cv":cur,"hw":hw}    → upd / size / md5 / delta
  └─ upd=0 → 无更新，结束
start    {"act":"start","product":P,"cv":cur,"trans":"F","chunk":r.chunk}
         → sid / size / total / md5
循环 pull（bin=1）直到 eof=1：
         {"act":"pull","sid":sid,"off":off,"batch":16,"bin":1}
         → cnt / next / eof  + cnt 个字节帧（按八、1 过滤）
校验     md5(产物) == 期望 md5          ← ★ 必须
finish   {"act":"finish","sid":sid,"result":"ok","md5":md5(产物)}
安装

start    {"act":"start","product":P,"cv":cur,"trans":"D"}
         → sid / trans="D" / size=<补丁大小> / md5=<目标完整包 MD5>
循环 pull → 得到 patch 字节
合成     new_img = bspatch(本地旧固件, patch)        ← BSDIFF40 格式
校验     md5(new_img) == start.md5                  ← ★ 必须，唯一护栏
finish   {"act":"finish","sid":sid,"result":"ok","md5":md5(new_img)}
安装

① 补丁下载失败 ────┐
② bspatch 合成失败 ─┼──► 回退全量下载（清理补丁与损坏产物）
③ 合成产物 MD5 不符 ┘         │
                        ├─ 成功 → 校验 MD5 → 安装
                        └─ 失败 → 最终失败

{"act":"changelog","product":P,"cv":cur}  →  {"ok":1,"content":"..."}

分片长度不符: 期望 8192 得到 0       ← 整包失败

int got = 0;
while (got < json.cnt) {
    f = recv_frame();
    if (f->frame_type != 0x00)      /* ★ 关键：HB_RESP / ERROR 等一律跳过 */
        continue;
    memcpy(buf + chunks[got].off, f->data, f->len);
    if (chunks[got].len != f->len)  /* 长度必须匹配，否则帧流已错位 */
        return ERROR;
    got++;
}

while (1) {
    r = pull(sid, off, batch, bin=1);
    if (r.again) { sleep_ms(50); continue; }   /* 退避重发，off 不变 */
    off = r.next;
    break;
}

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""TCP OTA 参考客户端 —— 完整实现了本文件的全部协议要求。"""

import struct, socket, json, time, hashlib, zlib, sys

SYNC      = b'\xAA\x55\x5A\xA5\xFF\xEC'
HDR_LEN   = 14
MAX_DATA  = 4 * 1024 * 1024

FT_DATA, FT_SESS_REQ, FT_SESS_RESP, FT_HB, FT_HB_RESP, FT_ERR = 0x00, 0x01, 0x02, 0x03, 0x04, 0x05
SVC_OTA = 1


# ── 帧编解码 ────────────────────────────────────────────────────────────────

def build_frame(ft, svc, sid, payload=b''):
    return SYNC + struct.pack('>BBHI', ft, svc, sid, len(payload)) + payload


class FrameReader:
    """按帧长度拆包；帧头损坏时重新找同步字。"""

    def __init__(self, sock):
        self.s = sock
        self.buf = bytearray()

    def _fill(self, n):
        while len(self.buf) < n:
            b = self.s.recv(65536)
            if not b:
                raise ConnectionError('连接已关闭')
            self.buf += b

    def read(self):
        """返回 (frame_type, svc_type, sid, payload)。"""
        while True:
            # 找同步字
            i = self.buf.find(SYNC)
            if i < 0:                       # 还没有完整同步字，丢多余的
                keep = len(SYNC) - 1
                del self.buf[:max(0, len(self.buf) - keep)]
                self._fill(len(SYNC))
                continue
            if i > 0:
                del self.buf[:i]            # 丢掉同步字之前的垃圾
            self._fill(HDR_LEN)
            ft, svc, sid, dlen = struct.unpack('>BBHI', self.buf[6:HDR_LEN])
            if dlen > MAX_DATA:             # 帧头损坏，跳一个字节重找
                del self.buf[:1]
                continue
            self._fill(HDR_LEN + dlen)
            payload = bytes(self.buf[HDR_LEN:HDR_LEN + dlen])
            del self.buf[:HDR_LEN + dlen]
            return ft, svc, sid, payload


class OtaClient:
    def __init__(self, host, port):
        self.sock = socket.create_connection((host, port), timeout=30)
        self.reader = FrameReader(self.sock)
        self.sid = 0
        self.last_hb = 0.0

    # ── 握手 ──
    def handshake(self):
        self.sock.sendall(build_frame(FT_SESS_REQ, SVC_OTA, 0))
        while True:
            ft, svc, sid, _ = self.reader.read()
            if ft == FT_SESS_RESP:
                self.sid = sid
                self.last_hb = time.time()
                return sid
            # 忽略握手前可能到达的其它帧

    def heartbeat(self, force=False):
        if force or time.time() - self.last_hb > 30:
            self.sock.sendall(build_frame(FT_HB, SVC_OTA, self.sid))
            self.last_hb = time.time()

    # ── 应用层请求：发一帧、收到一个 JSON 帧 ──
    def request(self, obj):
        self.heartbeat()
        payload = json.dumps(obj, ensure_ascii=False).encode('utf-8')
        self.sock.sendall(build_frame(FT_DATA, SVC_OTA, self.sid, payload))
        while True:
            ft, svc, sid, data = self.reader.read()
            if ft == FT_ERR:
                raise RuntimeError(f'链路错误码 0x{data[0]:02x}')
            if ft != FT_DATA:
                continue                     # ★ 跳过 HB_RESP 等插入帧
            return json.loads(data.decode('utf-8'))


# ── 高层流程 ────────────────────────────────────────────────────────────────

def do_check(c, product, cv, hw='', did=''):
    return c.request({'act': 'check', 'product': product, 'cv': cv,
                      'hw': hw, 'did': did})


def do_changelog(c, product, cv):
    r = c.request({'act': 'changelog', 'product': product, 'cv': cv})
    return r.get('content', '') if r.get('ok') else ''


def do_start(c, product, cv, trans=None, chunk=None, did=''):
    req = {'act': 'start', 'product': product, 'cv': cv, 'did': did}
    if trans:
        req['trans'] = trans
    if chunk:
        req['chunk'] = chunk
    return c.request(req)


def pull_all(c, sess_sid, total, batch=16, on_progress=None):
    """二进制模式拉取全部字节；处理 again 流控与插入帧。"""
    buf = bytearray(total)
    off = 0
    while off < total:
        while True:                                      # 处理 again
            r = c.request({'act': 'pull', 'sid': sess_sid, 'off': off,
                           'batch': batch, 'bin': 1})
            if not r.get('ok'):
                raise RuntimeError(f"pull 失败 err={r.get('err')} {r.get('msg','')}")
            if r.get('again'):
                time.sleep(0.05)                         # ★ 不退 off，退避重发
                continue
            break

        cnt = r['cnt']
        for i in range(cnt):
            # ★ 按帧类型过滤，只数 DATA 帧
            while True:
                ft, svc, sid, data = c.reader.read()
                if ft == FT_ERR:
                    raise RuntimeError(f'链路错误码 0x{data[0]:02x}')
                if ft == FT_DATA:
                    break
            meta = r['chunks'][i]
            if len(data) != meta['len']:
                raise RuntimeError(f"分片长度不符: 期望 {meta['len']} 得到 {len(data)}")
            # 可选：CRC32 校验
            crc_want = int(meta['crc'], 16)
            if zlib.crc32(data) & 0xFFFFFFFF != crc_want:
                raise RuntimeError(f"分片 CRC 不符 off={meta['off']}")
            buf[meta['off']:meta['off'] + meta['len']] = data

        off = r['next']
        if on_progress:
            on_progress(off, total)
        if r.get('eof'):
            break
    return bytes(buf)


def do_finish(c, sess_sid, result, md5):
    return c.request({'act': 'finish', 'sid': sess_sid,
                      'result': result, 'md5': md5})


def upgrade(host, port, product, cv, old_firmware=b'',
            hw='', did='', allow_delta=True):
    """完整升级流程（含差分优先 + 失败回退全量）。

    返回 (产物字节, 说明字符串)。异常表示升级失败。
    """
    c = OtaClient(host, port)
    c.handshake()
    c.request({'act': 'hello', 'chunk': 8192})

    info = do_check(c, product, cv, hw, did)
    if not info.get('ok'):
        raise RuntimeError(f"check 失败 err={info.get('err')} {info.get('msg','')}")
    if not info.get('upd'):
        return None, f"已是最新（busy={info.get('busy',0)}）"

    want_md5 = info['md5']
    chunk    = info.get('chunk', 8192)

    # ── 差分优先 ──
    if allow_delta and info.get('delta') and old_firmware:
        try:
            print(f"[差分] 补丁 {info['dsize']} 字节 "
                  f"（完整包的 {info['dsize']*100//info['size']}%）")
            s = do_start(c, product, cv, trans='D', chunk=chunk, did=did)
            if not s.get('ok'):
                raise RuntimeError(f"start(D) err={s.get('err')}")
            patch = pull_all(c, s['sid'], s['size'],
                             on_progress=lambda d, t: print(f"\r  下载 {d*100//t}%", end=''))
            print()

            import bsdiff4
            newimg = bsdiff4.patch(old_firmware, patch)     # ← 传**字节**，不是路径！
            got = hashlib.md5(newimg).hexdigest()
            if got != want_md5:
                raise RuntimeError(f"合成产物 MD5 不符: {got} != {want_md5}")
            do_finish(c, s['sid'], 'ok', got)
            return newimg, f"差分成功（补丁 {len(patch)} 字节，节省 "
        except Exception as e:
            print(f"[差分] 失败：{e} → 回退全量")             # ★ 必须回退

    # ── 全量 ──
    s = do_start(c, product, cv, trans='F', chunk=chunk, did=did)
    if not s.get('ok'):
        raise RuntimeError(f"start(F) err={s.get('err')}")
    img = pull_all(c, s['sid'], s['size'],
                   on_progress=lambda d, t: print(f"\r  下载 {d*100//t}%", end=''))
    print()
    got = hashlib.md5(img).hexdigest()
    if got != want_md5:
        do_finish(c, s['sid'], 'fail', got)
        raise RuntimeError(f"全量 MD5 不符: {got} != {want_md5}")
    do_finish(c, s['sid'], 'ok', got)
    return img, f"全量成功（{len(img)} 字节）"


if __name__ == '__main__':
    host = sys.argv[1] if len(sys.argv) > 1 else '127.0.0.1'
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8011
    prod = sys.argv[3] if len(sys.argv) > 3 else '喵账'
    ver  = sys.argv[4] if len(sys.argv) > 4 else '1.1.0'

    c = OtaClient(host, port)
    print('握手 sid =', c.handshake())
    print('hello   ', c.request({'act': 'hello', 'chunk': 8192}))
    print('check   ', do_check(c, prod, ver))
    print('changelog 前 80 字:', do_changelog(c, prod, ver)[:80])
    img, msg = upgrade(host, port, prod, ver)
    print('结果:', msg, f'MD5={hashlib.md5(img).hexdigest()}' if img else '')

python3 ota_client.py 127.0.0.1 8011 喵账 1.1.0        # 直连服务端口
python3 ota_client.py tcp.qftek.top 32200 喵账 1.1.0   # 走公网 + mux