STM32F4-Base/test/host/ftp_auth_bench.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""FTP 鉴权台架回归（需真机在线）

验证固件侧 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")
    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())