wandering1
d96113bcc4
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 比对， 建议纳入常备回归集）。
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
#!/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())
|