STM32F4-Base/MDK-ARM/ftp_freeze_probe.py

# -*- coding: utf-8 -*-
# Trustworthy TC-NET-604 freeze reproduction via the REAL production path.
#
# Why this one is trustworthy (unlike the 2026-09-22 probe batteries):
#   * Production firmware, no in-firmware readers/probes -> the "a test reader ate the
#     peer's bytes and the PC saw a short stream" artifact class cannot occur.
#   * Judgement = the FTP client's own protocol result (STOR/RETR replies) plus an
#     end-to-end SHA-256 taken from a RETR of the uploaded file (content ground truth).
#   * Data-channel stall vs whole-net-layer stall is distinguished by a NOOP on the
#     CONTROL channel (a different socket) while the data channel is stuck.
#   * Only the device's own production logs are used for the device-side timeline
#     (window them by the per-round "TS <epoch>" line printed here).
#
# Usage: ftp_freeze_probe.py [--ip <设备IP>] [--port 21] [--size 65536]
#                            [--rounds 10] [--gap 3] [--keep]
# Output lines (machine-auditable):
#   TS <epoch>
#   ROUND <n> size=<bytes> stor=<OK|FAIL:reason> t_stor=<s> retr=<OK|FAIL:reason>
#             t_retr=<s> hash=<MATCH|MISMATCH|NA> [ctrl=<alive|dead|na>]
#   SUMMARY rounds=<n> pass=<n> fail=<n>
# Exit: 0 all pass / 1 any failure / 2 usage or connect error.
import argparse, hashlib, io, os, random, socket, sys, time
from ftplib import FTP, error_reply, error_temp, error_perm, all_errors

# 台架连接默认值单一真源在 test/host/bench_defaults.py（跨目录：补 sys.path 后 import）
_sys_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                         "..", "test", "host")
if _sys_path not in sys.path:
    sys.path.insert(0, _sys_path)
import bench_defaults  # noqa: E402

# Device-side FTP auth: lftpd validates against /config.ini [ftp] (user / pwd_hash).
# This device has NO anonymous channel (USER anonymous -> 530), so the probe must log
# in with the factory account. Override with --user/--password if it was changed.
FTP_USER = "admin"
FTP_PASS = "admin"


def log(msg):
    sys.stdout.write(msg + "\n")
    sys.stdout.flush()

def liveness(ip, port, timeout=3.0):
    """Control-channel liveness on a FRESH connection (production lftpd control path)."""
    try:
        f = FTP()
        f.connect(ip, port, timeout=timeout)
        f.login(FTP_USER, FTP_PASS)
        f.voidcmd("NOOP")
        f.quit()
        return "alive"
    except Exception:
        return "dead"

def main():
    global FTP_USER, FTP_PASS   # 两处登录点都读模块常量，命令行覆盖后即刻生效
    ap = argparse.ArgumentParser()
    ap.add_argument("--ip", default=bench_defaults.DEVICE_IP)
    ap.add_argument("--port", type=int, default=21)
    ap.add_argument("--size", type=int, default=65536)
    ap.add_argument("--rounds", type=int, default=10)
    ap.add_argument("--gap", type=float, default=3.0)
    ap.add_argument("--keep", action="store_true", help="do not DELE the uploaded file")
    ap.add_argument("--seed", type=int, default=604)
    ap.add_argument("--user", default="admin",
                    help="FTP user (device auth; this device has no anonymous channel)")
    ap.add_argument("--password", default="admin", help="FTP password (factory default)")
    a = ap.parse_args()
    FTP_USER, FTP_PASS = a.user, a.password

    rng = random.Random(a.seed)
    npass = nfail = 0
    for n in range(1, a.rounds + 1):
        data = bytes(rng.getrandbits(8) for _ in range(a.size))
        want = hashlib.sha256(data).hexdigest()
        name = "/fz%04d.bin" % n
        log("TS %d" % int(time.time()))
        f = FTP()
        stor = retr = "NA"
        t_stor = t_retr = 0.0
        got_hash = "NA"
        ctrl = "na"
        try:
            f.connect(a.ip, a.port, timeout=15.0)
            f.login(FTP_USER, FTP_PASS)
            t0 = time.time()
            try:
                f.storbinary("STOR " + name, io.BytesIO(data), blocksize=8192)
                stor = "OK"
            except all_errors as e:
                stor = "FAIL:%s" % type(e).__name__
                ctrl = liveness(a.ip, a.port)
            t_stor = time.time() - t0
            if stor == "OK":
                buf = bytearray()
                t0 = time.time()
                try:
                    f.retrbinary("RETR " + name, buf.extend, blocksize=8192)
                    retr = "OK"
                    got_hash = "MATCH" if hashlib.sha256(bytes(buf)).hexdigest() == want else "MISMATCH"
                except all_errors as e:
                    retr = "FAIL:%s" % type(e).__name__
                    ctrl = liveness(a.ip, a.port)
                t_retr = time.time() - t0
                if not a.keep:
                    try:
                        f.delete(name)
                    except all_errors:
                        pass
            try:
                f.quit()
            except all_errors:
                try:
                    f.close()
                except Exception:
                    pass
        except all_errors as e:
            stor = "FAIL:connect-%s" % type(e).__name__
            ctrl = liveness(a.ip, a.port)
        ok = (stor == "OK" and retr == "OK" and got_hash == "MATCH")
        npass += 1 if ok else 0
        nfail += 0 if ok else 1
        log("ROUND %d size=%d stor=%s t_stor=%.2f retr=%s t_retr=%.2f hash=%s ctrl=%s" %
            (n, a.size, stor, t_stor, retr, t_retr, got_hash, ctrl))
        if a.gap > 0 and n < a.rounds:
            time.sleep(a.gap)
    log("SUMMARY rounds=%d pass=%d fail=%d" % (a.rounds, npass, nfail))
    return 0 if nfail == 0 else 1

sys.exit(main())