W5波+N-19收口：FTP鉴权落地、Trap25三连定案修复(N-30)、N-20/N-23/N-26
N-19 FTP 鉴权：
- lftpd_auth（USER/PASS 状态机 + 连续5次失败锁60s）+ SHA-256
(user||pwd) 口令哈希；config.ini [ftp] 段 user/pwd_hash/write，
出厂 admin/admin 在用有启动告警；lftpd_log.c 删除并入 lftpd_auth
- ftp_cfg 配置读写；台架 ftp_auth_bench.py 13/13
N-30 / Trap 25（CH395F_Trap_Records.md 四次定案全记录）：
- 修复1 closed-with-data accept：客户端发完正常 FIN 后芯片按
FUN_PARA(0x09) §5.52② 保留 Socket+RX 缓冲，旧代码把 CLOSED 一律
relisten 自毁缓冲 → STOR 必挂 425；改为接受"已关但可读"连接排空
- 修复2 丢 CONNECT 中断探针 net_lost_connect_probe + last_peer_port
幽灵甄别（拆链窗口握手完成但中断丢失，双信源去抖补 accept）
- 修复3 send_list/send_nlst 批量单发：背靠背 ~60B 小段毒化芯片 TX
（芯片报 SEND_OK 但对端全丢），整份目录一次性写入
- 证伪存档：固定沉降/Socket 轮换冷却/listen 门控（台架 v7 矩阵）
- 台架工具 test/board/ch395f_reuse_bench.{c,h} + host 脚本存档
（不注册编译；实验分支 n30-chip-bench 保留）
N-20：Error_Handler 死循环降级（记录+复位路径）
N-23：scripts/check_registration.py 双端注册反向校验（pre-commit）
N-26：GD5F NAND 直 DMA 调用方缓冲窗口断言
验证：BUILD_OK 0/0；ftp_regression 连跑 4 轮 11/11（修复前
LIST#4/#7 100% 挂、STOR#2 曾 100% 425）；ftp_auth_bench 13/13；
@test 启动标记 PASS；ci_local L0+L0.5 全过；doc-links 通过。
登记簿 N-30 已销项。
This commit is contained in:
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""FTP 鉴权台架回归（N-19，需真机在线）
|
||||
|
||||
验证固件侧 lftpd 鉴权集成的完整行为矩阵（主机单测 test/host/ftp_auth 覆盖
|
||||
不到的"协议接线"部分：命令门、530/331/230/550 响应码、跨会话锁定）：
|
||||
|
||||
T1 匿名登录（anonymous） → 期望 530 拒
|
||||
T2 未登录直接 NLST → 期望 530（命令门）
|
||||
T3 对用户名 + 错口令 ×5 → 期望 530 ×5（并触发锁定）
|
||||
T4 锁定窗内正确口令 → 期望 530（LOCKED，不算哈希）
|
||||
T5 等待锁定期满后正确口令 → 期望 230（60s + 余量）
|
||||
T6 登录后 NLST / STOR / RETR / DELE → 期望全通（出厂 write=1）
|
||||
T7 会话结束重连、错口令一次 → 期望 530 且计数已清零（T5 成功清零）
|
||||
|
||||
注意：CH395F 单连接——每步用完必须 quit() 释放 socket，否则下一步连不上；
|
||||
跑之前确保 UI 远程显示(7003)/对时(8888)客户端未占用 socket。
|
||||
|
||||
用法（设备默认 192.168.1.100，出厂口令 admin/admin）：
|
||||
& '.venv\\Scripts\\python.exe' test/host/ftp_auth_bench.py
|
||||
改过口令： ... --pwd <新口令>
|
||||
跳过 60s 锁定等待（只验锁定触发不验解锁）： ... --skip-unlock
|
||||
|
||||
退出码：0=全部通过 / 1=任一失败 / 2=环境错误（连不上设备）
|
||||
创建日期：2026-10-11
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import io
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
from ftplib import FTP, error_perm
|
||||
|
||||
LOCK_WAIT_S = 62 # 固件锁定 60 s + 余量
|
||||
|
||||
|
||||
def _try_login(host, port, timeout, user, pwd):
|
||||
"""尝试一次登录；返回 (ok, code_or_exc)。失败不断言连接状态。
|
||||
|
||||
单连接 socket：上一次会话结束后 net 层 auto_relisten ~50 ms，但 accept
|
||||
失败会走"毁监听→重建"路径（3 s 重试窗）→ connect 被拒时重试 3 次。
|
||||
"""
|
||||
f = FTP()
|
||||
last_err = None
|
||||
for attempt in range(3):
|
||||
try:
|
||||
time.sleep(0.5)
|
||||
f.connect(host, port, timeout=timeout)
|
||||
break
|
||||
except (ConnectionRefusedError, socket.timeout, OSError) as e:
|
||||
last_err = e
|
||||
f = FTP()
|
||||
else:
|
||||
return None, repr(last_err), None
|
||||
try:
|
||||
try:
|
||||
f.login(user, pwd)
|
||||
return True, "230", f
|
||||
except error_perm as e:
|
||||
f.close()
|
||||
return False, str(e), None
|
||||
except (socket.timeout, OSError) as e:
|
||||
return None, repr(e), None
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(description="FTP auth bench matrix (N-19)")
|
||||
ap.add_argument("--host", default="192.168.1.100")
|
||||
ap.add_argument("--port", type=int, default=21)
|
||||
ap.add_argument("--timeout", type=int, default=15)
|
||||
ap.add_argument("--user", default="admin")
|
||||
ap.add_argument("--pwd", default="admin")
|
||||
ap.add_argument("--skip-unlock", action="store_true",
|
||||
help="不验 60s 解锁（省 1 分钟；锁定触发仍验）")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
fails = []
|
||||
|
||||
def expect(name, cond, detail=""):
|
||||
tag = "PASS" if cond else "FAIL"
|
||||
print(" [%s] %s %s" % (tag, name, detail))
|
||||
if not cond:
|
||||
fails.append(name)
|
||||
|
||||
# ---- T1 匿名登录被拒 ----
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout, "anonymous", "x@x")
|
||||
if ok is None:
|
||||
print("[env] 连不上 %s:%d → %s" % (args.host, args.port, code))
|
||||
return 2
|
||||
expect("T1 匿名登录 → 530", (ok is False) and code.startswith("530"), code)
|
||||
if conn:
|
||||
conn.quit()
|
||||
|
||||
# ---- T2 未登录直接 NLST → 530（命令门）----
|
||||
time.sleep(0.5) # 单连接 socket：等 auto_relisten 沉降（~50 ms + 余量）
|
||||
f = FTP()
|
||||
f.connect(args.host, args.port, timeout=args.timeout)
|
||||
try:
|
||||
f.sendcmd("NLST")
|
||||
expect("T2 未登录 NLST → 530", False, "居然放行")
|
||||
except error_perm as e:
|
||||
expect("T2 未登录 NLST → 530", str(e).startswith("530"), str(e))
|
||||
except Exception as e: # noqa: BLE001
|
||||
expect("T2 未登录 NLST → 530", False, repr(e))
|
||||
# ⚠ 必须 close：单连接模型下不关 = 占住 socket，后续步骤全被拒
|
||||
# （2026-10-11 台架实测：QUIT 后不 close → 芯片视角旧连接仍 ESTABLISHED）
|
||||
try:
|
||||
f.quit()
|
||||
except Exception: # noqa: BLE001
|
||||
try:
|
||||
f.close()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
time.sleep(0.5)
|
||||
|
||||
# ---- T3 错口令 ×5 → 530 ×5 ----
|
||||
for i in range(1, 6):
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout,
|
||||
args.user, "wrong-pass-%d" % i)
|
||||
expect("T3.%d 错口令 → 530" % i, (ok is False) and code.startswith("530"), code)
|
||||
if conn:
|
||||
conn.quit()
|
||||
|
||||
# ---- T4 锁定窗内正确口令 → 530 ----
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout, args.user, args.pwd)
|
||||
expect("T4 锁定窗内对口令 → 530", (ok is False) and code.startswith("530"), code)
|
||||
if conn:
|
||||
conn.quit()
|
||||
|
||||
if args.skip_unlock:
|
||||
print("[skip] 未验 60s 解锁（--skip-unlock）")
|
||||
else:
|
||||
# ---- T5 期满正确口令 → 230 ----
|
||||
print(" [wait] 等锁定期满 %d s ..." % LOCK_WAIT_S)
|
||||
time.sleep(LOCK_WAIT_S)
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout,
|
||||
args.user, args.pwd)
|
||||
expect("T5 期满对口令 → 230", ok is True, code)
|
||||
if conn:
|
||||
# ---- T6 登录后读写全通（出厂 write=1）----
|
||||
try:
|
||||
names = conn.nlst()
|
||||
expect("T6a NLST", len(names) >= 0, "%d 项" % len(names))
|
||||
payload = b"N19-auth-bench-" + b"X" * 500
|
||||
conn.storbinary("STOR n19t.tmp", io.BytesIO(payload))
|
||||
got = io.BytesIO()
|
||||
conn.retrbinary("RETR n19t.tmp", got.write)
|
||||
expect("T6b STOR/RETR 往返一致", got.getvalue() == payload)
|
||||
conn.delete("n19t.tmp")
|
||||
expect("T6c DELE", True)
|
||||
except Exception as e: # noqa: BLE001
|
||||
expect("T6 登录后读写", False, repr(e))
|
||||
# ---- T7 成功已清零计数：本会话退出后错一次只算 1（不锁）----
|
||||
conn.quit()
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout,
|
||||
args.user, "one-wrong")
|
||||
expect("T7 成功后错 1 次 → 530 且未锁（下次对口令可登）",
|
||||
(ok is False) and code.startswith("530"), code)
|
||||
if conn:
|
||||
conn.quit()
|
||||
ok, code, conn = _try_login(args.host, args.port, args.timeout,
|
||||
args.user, args.pwd)
|
||||
expect("T7b 紧接对口令 → 230", ok is True, code)
|
||||
if conn:
|
||||
conn.quit()
|
||||
|
||||
print("\nVERDICT: %s (%d fail)" % ("PASS" if not fails else "FAIL", len(fails)))
|
||||
return 0 if not fails else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user