112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
# -*- 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 192.168.1.100] [--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
|
|
|
|
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("anonymous", "probe@local")
|
|
f.voidcmd("NOOP")
|
|
f.quit()
|
|
return "alive"
|
|
except Exception:
|
|
return "dead"
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--ip", default="192.168.1.100")
|
|
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)
|
|
a = ap.parse_args()
|
|
|
|
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("anonymous", "probe@local")
|
|
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())
|