docs(net): N34 E11/E12 实验记录与台架脚本（KeepAlive 服务器侧缺口、TSL 规格偏离）

E11：服务器侧 KeepAlive 未开启（静态证据链 + 注入手段两次修正 + 实测生产参数 30.0s 判死、空闲存活会话不误杀），给出采纳建议；防火墙规则对已建立连接无效、有效注入 = 禁用 PC 网卡，两条方法学留痕。 E12：缓存区读写路径长度→数据零间隔属规格偏离（读 0.4us/写 0.2us），但 460 轮/约 58MB 字节级校验未测得可复现危害；依 Trap 25 登记偏离、 B 变体（两段+TSL，+28B，吞吐无回退）留作待用补丁。残渣风暴态跨烧录存活污染 A/B 连接级对比，教训写入 E12：实验前后必须冷启动（断电重上电），MCU 覆写烧录/RESET_ALL 不足以清除。新增 test/host/e11_keepalive_probe.py（idle/block 两模式 + 注入正对照）与 test/host/ftp_integrity_test.py（STOR→RETR 全量 SHA-256 比对，建议纳入常备回归集）。
2026-10-13 09:10:17 +08:00
parent eac36526e7
commit d96113bcc4
4 changed files with 418 additions and 0 deletions
@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""E11：CH395F 服务器侧 KeepAlive —— "链路在、对端静默"判死验证（PC 侧注入）。
+
+背景（N34 E11）：手册 §5.56 规定 TCP 服务器应在 SINT_STAT_CONNECT 后开启
+KeepAlive；本仓该调用 2026-10-12 被撤回，全仓仅剩客户端分支启用，而
+App/ 无任何 net_connect() → 产品上 keepalive 实际关闭。本脚本用于验证
+"开启后能否在有界时间内判死静默对端，且不误杀正常空闲会话"。
+
+用法：
+  # T2 正常空闲存活：连接后静默 hold 秒，末尾发 KEEPLIVE 验证会话仍在
+  python e11_keepalive_probe.py --mode idle --hold-s 90
+
+  # T1 静默对端：连接后立即装 Windows 防火墙入站 DROP（仅本连接本地端口 +
+  #    设备 IP）→ 芯片 keepalive 探测得不到 ACK → 期望设备在
+  #    IDLE + INTVL×CNT 内判死（本仓 60 + 5x3 = 75s）
+  python e11_keepalive_probe.py --mode block --hold-s 120
+
+判据（设备侧事件由串口日志核对，本脚本打印 PC 侧时间线）：
+  T1 期望：串口出现 `[ERR][NET] sockN TIMEOUT`（耗时≈75s），随后
+           `sockN pump/relisten` 恢复；本脚本移除规则后新连接必须成功。
+  T2 期望：hold 期间**无** TIMEOUT；末尾 KEEPLIVE 必须收到 0xBB 回显。
+
+需要管理员权限（仅 block 模式装/卸防火墙规则；脚本自行清理）。
+"""
+
+import argparse
+import socket
+import subprocess
+import sys
+import time
+
+TAG_C, TAG_D = 0xAA, 0xBB
+CMD_KEEPLIVE = 0
+RULE_NAME = "N34-E11-DROP"
+
+
+def _crc(data: bytes) -> int:
+    c = 0
+    for b in data:
+        c ^= b
+    return c
+
+
+def build_frame(cmd: int, data: bytes = b"") -> bytes:
+    n = len(data)
+    return bytes([TAG_C, cmd, (n >> 8) & 0xFF, n & 0xFF]) + data + bytes([_crc(data)])
+
+
+def recv_frame(sock: socket.socket, timeout: float = 3.0):
+    sock.settimeout(timeout)
+    hdr = b""
+    while len(hdr) < 4:
+        chunk = sock.recv(4 - len(hdr))
+        if not chunk:
+            return None
+        hdr += chunk
+    tag, cmd, n = hdr[0], hdr[1], (hdr[2] << 8) | hdr[3]
+    body = b""
+    while len(body) < n + 1:
+        chunk = sock.recv(n + 1 - len(body))
+        if not chunk:
+            return None
+        body += chunk
+    data, crcb = body[:n], body[n]
+    return cmd, data, (tag == TAG_D) and (crcb == _crc(data))
+
+
+def firewall(action: str, local_port: int, remote_ip: str) -> bool:
+    verb = "add" if action == "add" else "delete"
+    cmd = ["netsh", "advfirewall", "firewall", verb, "rule",
+           f"name={RULE_NAME}", "dir=in", "action=block", "protocol=TCP",
+           f"localport={local_port}", f"remoteip={remote_ip}"]
+    if verb == "delete":
+        cmd = ["netsh", "advfirewall", "firewall", "delete", "rule",
+               f"name={RULE_NAME}"]
+    r = subprocess.run(cmd, capture_output=True, text=True)
+    if r.returncode != 0:
+        print(f"[WARN] 防火墙 {verb} 失败：{r.stdout.strip()} {r.stderr.strip()}")
+        return False
+    return True
+
+
+def connect(host: str, port: int, timeout: float = 5.0):
+    s = socket.create_connection((host, port), timeout)
+    return s
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="192.168.1.100")
+    ap.add_argument("--port", type=int, default=7003)
+    ap.add_argument("--mode", choices=["idle", "block"], required=True)
+    ap.add_argument("--hold-s", type=float, default=90.0)
+    ap.add_argument("--expect-detect-s", type=float, default=75.0,
+                    help="判死上界（IDLE + INTVL*CNT），仅用于提示")
+    args = ap.parse_args()
+
+    t0 = time.time()
+
+    def log(msg: str) -> None:
+        print(f"[{time.strftime('%H:%M:%S')} +{time.time() - t0:6.1f}s] {msg}",
+              flush=True)
+
+    log(f"模式={args.mode} 目标={args.host}:{args.port} hold={args.hold_s}s")
+    log(f"（设备侧判死上界参考 = {args.expect_detect_s}s）")
+
+    try:
+        s = connect(args.host, args.port)
+    except OSError as e:
+        print(f"[FAIL] 连接失败：{e}")
+        return 2
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 0)  # PC 侧不参与
+    lport = s.getsockname()[1]
+    log(f"已连接（本地端口 {lport}）→ 进入静默")
+
+    rc = 0
+    if args.mode == "block":
+        # 入站 DROP：设备发来的 keepalive 探测包在 PC 侧被丢弃 → 永不回 ACK
+        if not firewall("add", lport, args.host):
+            s.close()
+            return 2
+        log(f"已装防火墙入站 DROP（本地端口 {lport} <- {args.host}）→ 对端静默开始")
+
+    try:
+        time.sleep(args.hold_s)
+    except KeyboardInterrupt:
+        pass
+
+    if args.mode == "idle":
+        # 存活探针：KEEPLIVE 必须被回显
+        log("静默结束 → 发 KEEPLIVE 验证会话存活")
+        try:
+            s.sendall(build_frame(CMD_KEEPLIVE))
+            r = recv_frame(s, 3.0)
+        except OSError as e:
+            r = None
+            log(f"KEEPLIVE 发送/接收异常：{e}")
+        if r and r[2]:
+            log("PASS：收到 0xBB KEEPLIVE 回显 → 正常空闲会话未被误杀")
+        else:
+            log(f"FAIL：未收到有效回显（{r}）→ 空闲置的会话被断开（误杀）")
+            rc = 1
+        s.close()
+        return rc
+
+    # block 模式：卸规则 → 验证旧连接已死 + 监听器已恢复
+    log("静默结束 → 卸载防火墙规则")
+    firewall("delete", lport, args.host)
+    time.sleep(1.0)
+
+    # 旧连接是否已被设备判死（应读到 EOF/RST；读不到也算进程侧正常）
+    try:
+        s.settimeout(2.0)
+        d = s.recv(16)
+        log(f"旧连接 recv → {d!r}（b'' = 已收到 FIN，即设备已判死并释放）")
+    except OSError as e:
+        log(f"旧连接 recv 异常（设备已判死/连接被拆）：{e}")
+    s.close()
+
+    # 监听器恢复验证：新连接 + KEEPLIVE 回显
+    log("验证监听器恢复：新建连接 + KEEPLIVE")
+    ok = False
+    for attempt in range(1, 6):
+        try:
+            s2 = connect(args.host, args.port)
+            s2.sendall(build_frame(CMD_KEEPLIVE))
+            r = recv_frame(s2, 3.0)
+            s2.close()
+            if r and r[2]:
+                log(f"PASS：第 {attempt} 次新连接建立并收到回显 → 监听器已恢复")
+                ok = True
+                break
+        except OSError as e:
+            log(f"第 {attempt} 次新连接失败：{e}")
+        time.sleep(2.0)
+    if not ok:
+        log("FAIL：监听器未恢复（端口仍不接客）")
+        rc = 1
+    return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""E12：CH395F SPI 缓存区 TSL 时序 A/B 的字节级校验工作负载。
+
+背景（N34 E12）：手册 §7.4 规定读写收发缓存区命令需在"长度与数据之间"留
+TSL（读 0.4us / 写 0.2us）。本仓 ch395f_write_send_buf / ch395f_read_recv_buf
+把命令头与数据打包进**同一 DMA 突发**（零间隔）。本脚本用 SHA-256 逐轮校验
+"上传→回读"是否字节一致，为 A/B 提供灵敏判据：
+  - STOR（PC→设备）走设备 read_recv_buf（TSL 读 0.4us）
+  - RETR（设备→PC）走设备 write_send_buf（TSL 写 0.2us）
+任一方向出现比特错位都会让 SHA 不匹配。
+
+用法：
+  python ftp_integrity_test.py --rounds 20 --size 65536
+  python ftp_integrity_test.py --rounds 20 --size 65536 --tag A   # 结果标记
+
+判据：全部轮次 SHA-256 一致 = PASS；出现不一致则打印首个差异偏移，
+并自动二次回读以区分"存储副本错（写路径/上传）"与"回读瞬时错（读路径）"。
+"""
+
+import argparse
+import hashlib
+import io
+import random
+import sys
+import time
+from ftplib import FTP, error_perm
+
+NAME = "e12a.bin"          # 8.3 短名（FatFs FF_USE_LFN=0）
+
+
+def sha(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+
+def first_diff(a: bytes, b: bytes):
+    n = min(len(a), len(b))
+    for i in range(n):
+        if a[i] != b[i]:
+            return i, a[i], b[i]
+    if len(a) != len(b):
+        return n, None, None
+    return None
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="192.168.1.100")
+    ap.add_argument("--port", type=int, default=21)
+    ap.add_argument("--user", default="admin")
+    ap.add_argument("--pwd", default="admin")
+    ap.add_argument("--rounds", type=int, default=20)
+    ap.add_argument("--size", type=int, default=65536, help="每轮载荷字节数")
+    ap.add_argument("--seed", type=int, default=20261013)
+    ap.add_argument("--tag", default="", help="结果标记（A/B 分组）")
+    args = ap.parse_args()
+
+    rng = random.Random(args.seed)
+    ftp = FTP()
+    ftp.connect(args.host, args.port, timeout=20)
+    ftp.login(args.user, args.pwd)
+    ftp.set_pasv(True)
+    print(f"[E12{args.tag}] 已登录 {args.host}:{args.port} "
+          f"rounds={args.rounds} size={args.size}B", flush=True)
+
+    bad = 0
+    t_all = time.time()
+    for i in range(1, args.rounds + 1):
+        payload = rng.randbytes(args.size)
+        want = sha(payload)
+
+        t0 = time.time()
+        ftp.storbinary(f"STOR {NAME}", io.BytesIO(payload))
+
+        got = bytearray()
+        ftp.retrbinary(f"RETR {NAME}", got.extend)
+        got = bytes(got)
+        dt = time.time() - t0
+
+        if got == payload:
+            print(f"  round {i:2d}/{args.rounds} OK   "
+                  f"size={len(got)}B sha={want[:16]} {dt:.2f}s "
+                  f"({args.size / 1024 / max(dt, 1e-3):.0f} KB/s)", flush=True)
+            continue
+
+        bad += 1
+        fd = first_diff(got, payload)
+        print(f"  round {i:2d}/{args.rounds} MISMATCH len={len(got)} want={len(payload)} "
+              f"first_diff={fd}", flush=True)
+        # 归因：二次回读
+        got2 = bytearray()
+        try:
+            ftp.retrbinary(f"RETR {NAME}", got2.extend)
+            got2 = bytes(got2)
+            if got2 == payload:
+                print("       → 二次回读一致：判为回读瞬时错（read 路径）", flush=True)
+            else:
+                print(f"       → 二次回读仍不一致（sha2={sha(got2)[:16]}）："
+                      f"判为存储副本错（write/上传路径）", flush=True)
+        except error_perm as e:
+            print(f"       → 二次回读失败：{e}", flush=True)
+
+    ftp.quit() if False else ftp.close()
+    total = time.time() - t_all
+    print(f"[E12{args.tag}] 汇总：{args.rounds - bad}/{args.rounds} 轮字节一致"
+          f"（{bad} 轮不一致），总耗时 {total:.1f}s", flush=True)
+    return 1 if bad else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())