406 lines
16 KiB
Python
406 lines
16 KiB
Python
|
"""lan_probe.py — 局域网设备探测原语（无状态：一次请求跑一批，跑完就返回）。
|
|||
|
|
|
|||
|
|
职责切分：候选表、分批、进度、取消、去重、渲染、结果缓存全在页面
|
|||
|
|
(js/common/lan-picker.js)；本模块只做两件事——
|
|||
|
|
probe_batch(ips, ports, path, timeout_ms) 逐 IP 一次 HTTP GET + 判决
|
|||
|
|
local_net() 本机默认路由源 IP / 派生 /24
|
|||
|
|
|
|||
|
|
为什么不用 urllib/requests：Windows 上它们的 connect 超时和 read 超时共用一个
|
|||
|
|
预算，冷 ARP + WiFi 省电唤醒能让一台活板子整批假死；同仓实测记录见
|
|||
|
|
ecat_client.py:319-335。这里用单个 selectors 循环驱动全部非阻塞 socket，
|
|||
|
|
chunk 内所有目标共用一个绝对截止，最坏耗时 = ceil(n/concurrency) * timeout。
|
|||
|
|
|
|||
|
|
判决三态（不许有未定义分支，兜底一律 unknown）：
|
|||
|
|
board — 实现了 /system/status?lite=1 且 wifi/gd32_features 形状对得上
|
|||
|
|
rejected — 有**反证**（非 200 / 非 JSON / 无 wifi 对象 / STA 自报 IP 与探测地址不符）
|
|||
|
|
unknown — 连不上、读超时、或形状合但落不进上面两类
|
|||
|
|
把"没测出来"塞进 rejected 就是自证判据；本模块宁可留灰也不猜。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import selectors
|
|||
|
|
import socket
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
# 探测路径固定 lite：256 B 栈缓冲、不抢 httpJsonWorkMutex、不发 SPI 事务
|
|||
|
|
# (ESP32_COMM/main/server/http_handlers.cpp:1769-1806)。
|
|||
|
|
PROBE_PATH = "/system/status?lite=1"
|
|||
|
|
# 白名单而不是随便传：/system/probe 会发起真实 SPI 事务，/logs 会排空固件的
|
|||
|
|
# 日志环形缓冲——扫 254 台时这些都会污染正在调试的东西。要加路径先去固件仓确认副作用。
|
|||
|
|
ALLOWED_PATHS = (PROBE_PATH, "/system/status")
|
|||
|
|
|
|||
|
|
BATCH_MAX_IPS = 64 # 一次请求最多探多少个 IP（页面按此分批）
|
|||
|
|
# 并发是"同时探多少个不同地址"，不是对同一个地址开多条连接——所以一块板子
|
|||
|
|
# 在整场扫描里最多只会被我们占 1 个 socket（它 httpd 只有 6 个且开 lru_purge）。
|
|||
|
|
# 整段耗时由"不存在的地址等满截止"主导：254 个地址在 8 并发下要 24s，48 并发约 4s。
|
|||
|
|
CONCURRENCY = 48
|
|||
|
|
TIMEOUT_MS_MAX = 3000 # 单目标硬截止上限
|
|||
|
|
BUDGET_MS_MAX = 8000 # 一批的最长总耗时（超出部分回 pending，不谎报探过）
|
|||
|
|
HEAD_BYTES = 96 # rejected 里带回的响应片段长度（给人念出来用）
|
|||
|
|
|
|||
|
|
|
|||
|
|
def valid_ipv4(s):
|
|||
|
|
"""严格 IPv4 点分十进制校验：四段 0-255、纯数字、不允许前导多余字符。"""
|
|||
|
|
parts = str(s or "").strip().split(".")
|
|||
|
|
if len(parts) != 4:
|
|||
|
|
return False
|
|||
|
|
for p in parts:
|
|||
|
|
if not p.isdigit() or len(p) > 3 or not 0 <= int(p) <= 255:
|
|||
|
|
return False
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
|
|||
|
|
def local_net():
|
|||
|
|
"""默认路由源 IP + 派生 /24 + 本机全部 IPv4（供页面过滤自己的地址）。
|
|||
|
|
|
|||
|
|
UDP connect 只做内核路由查询，不发一个包。取不到默认路由就返回 ok=False，
|
|||
|
|
页面必须要求人填 CIDR —— 不回退成"那就打 .127 试一把"，那是配置自证。
|
|||
|
|
"""
|
|||
|
|
for dst in (("8.8.8.8", 53), ("223.5.5.5", 53), ("1.1.1.1", 53)):
|
|||
|
|
ip = None
|
|||
|
|
try:
|
|||
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|||
|
|
try:
|
|||
|
|
s.connect(dst)
|
|||
|
|
ip = s.getsockname()[0]
|
|||
|
|
finally:
|
|||
|
|
s.close()
|
|||
|
|
except OSError:
|
|||
|
|
continue
|
|||
|
|
if not ip or ip.startswith("169.254.") or ip.startswith("127."):
|
|||
|
|
continue
|
|||
|
|
return {
|
|||
|
|
"ok": True,
|
|||
|
|
"default_src_ip": ip,
|
|||
|
|
"cidr": ".".join(ip.split(".")[:3]) + ".0/24",
|
|||
|
|
"source": "udp-route",
|
|||
|
|
"local_ips": _all_local_ipv4(),
|
|||
|
|
}
|
|||
|
|
return {"ok": False, "err": "no-default-route", "local_ips": _all_local_ipv4()}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _all_local_ipv4():
|
|||
|
|
out = []
|
|||
|
|
try:
|
|||
|
|
for ai in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
|
|||
|
|
ip = ai[4][0]
|
|||
|
|
if ip not in out:
|
|||
|
|
out.append(ip)
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
def classify(ip, status, ctype, body):
|
|||
|
|
"""纯判决函数：给一次 HTTP 响应，返回该地址是不是我们的板子。
|
|||
|
|
|
|||
|
|
单独拎出来是为了能被离线单测打（scripts/test_lan_probe.py），不碰 socket。
|
|||
|
|
"""
|
|||
|
|
if status is None:
|
|||
|
|
return {"verdict": "unknown", "reason": "no-response"}
|
|||
|
|
if status != 200:
|
|||
|
|
return {"verdict": "rejected", "reason": "http-%d" % status}
|
|||
|
|
if "json" not in (ctype or "").lower():
|
|||
|
|
# 路由器/NAS 登录页落这里
|
|||
|
|
return {"verdict": "rejected", "reason": "not-json"}
|
|||
|
|
try:
|
|||
|
|
j = json.loads(body)
|
|||
|
|
except (ValueError, TypeError):
|
|||
|
|
return {"verdict": "rejected", "reason": "bad-json"}
|
|||
|
|
if not isinstance(j, dict):
|
|||
|
|
return {"verdict": "rejected", "reason": "not-object"}
|
|||
|
|
|
|||
|
|
wifi = j.get("wifi")
|
|||
|
|
if not isinstance(wifi, dict):
|
|||
|
|
return {"verdict": "rejected", "reason": "no-wifi-object"}
|
|||
|
|
mode = wifi.get("mode")
|
|||
|
|
if mode not in ("STA", "AP"):
|
|||
|
|
# 形状不对但不是反证：固件字段变了/别的 JSON 服务都落这里
|
|||
|
|
return {"verdict": "unknown", "reason": "bad-mode"}
|
|||
|
|
|
|||
|
|
gf = j.get("gd32_features")
|
|||
|
|
schema = "core-only"
|
|||
|
|
if isinstance(gf, dict) and isinstance(gf.get("known"), bool) \
|
|||
|
|
and isinstance(gf.get("flags"), int):
|
|||
|
|
schema = "full"
|
|||
|
|
|
|||
|
|
self_ip = wifi.get("ip")
|
|||
|
|
connected = wifi.get("connected")
|
|||
|
|
|
|||
|
|
if mode == "AP" and connected is not True:
|
|||
|
|
# 关键坑：lite 在 NOT(STA && connected) 时一律报 mode:"AP"
|
|||
|
|
# (http_handlers.cpp:1776-1789)。所以 AP+connected:false 多半是
|
|||
|
|
# "STA 掉租约的板子"，既不能当在线 AP 板收下，也不能当反证扔掉。
|
|||
|
|
return {"verdict": "unknown", "reason": "ap-not-connected",
|
|||
|
|
"schema": schema, "mode": mode, "self_ip": self_ip}
|
|||
|
|
|
|||
|
|
if mode == "STA" and self_ip not in (ip, None, ""):
|
|||
|
|
# 反证：它自称在别处。NAT/端口转发/捕获门户落这里。
|
|||
|
|
return {"verdict": "rejected", "reason": "ip-mismatch:%s" % self_ip,
|
|||
|
|
"schema": schema}
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"verdict": "board",
|
|||
|
|
"kind": "ap" if mode == "AP" else "sta",
|
|||
|
|
"schema": schema,
|
|||
|
|
"mode": mode,
|
|||
|
|
"connected": bool(connected),
|
|||
|
|
"self_ip": self_ip,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_http(buf):
|
|||
|
|
"""从原始字节里取 (status, content_type, body)。不引 http.client：只要这三样。"""
|
|||
|
|
sep = buf.find(b"\r\n\r\n")
|
|||
|
|
head = buf[:sep if sep >= 0 else len(buf)]
|
|||
|
|
body = buf[sep + 4:] if sep >= 0 else b""
|
|||
|
|
lines = head.split(b"\r\n")
|
|||
|
|
status = None
|
|||
|
|
if lines and lines[0]:
|
|||
|
|
bits = lines[0].split()
|
|||
|
|
if len(bits) >= 2 and bits[1].isdigit():
|
|||
|
|
status = int(bits[1])
|
|||
|
|
ctype = ""
|
|||
|
|
for ln in lines[1:]:
|
|||
|
|
k, _, v = ln.partition(b":")
|
|||
|
|
if k.lower() == b"content-type":
|
|||
|
|
ctype = v.decode("latin-1", "replace").strip()
|
|||
|
|
break
|
|||
|
|
return status, ctype, body.decode("utf-8", "replace")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _content_length(head_bytes_buf):
|
|||
|
|
i = head_bytes_buf.find(b"\r\n\r\n")
|
|||
|
|
if i < 0:
|
|||
|
|
return None
|
|||
|
|
for ln in head_bytes_buf[:i].split(b"\r\n")[1:]:
|
|||
|
|
k, _, v = ln.partition(b":")
|
|||
|
|
if k.lower() == b"content-length":
|
|||
|
|
try:
|
|||
|
|
return int(v.strip())
|
|||
|
|
except ValueError:
|
|||
|
|
return None
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _probe_chunk(targets, path, deadline, errors):
|
|||
|
|
"""并发探完一小批（≤CONCURRENCY）。返回与 targets 同序的结果列表。
|
|||
|
|
|
|||
|
|
每个 socket 都登记在 active 里，退出路径只有一条（close + pop），所以句柄
|
|||
|
|
不会随"页面中途关标签、这批没人收"而泄漏。
|
|||
|
|
"""
|
|||
|
|
sel = selectors.DefaultSelector()
|
|||
|
|
states = []
|
|||
|
|
active = {} # sock -> state
|
|||
|
|
|
|||
|
|
def retire(sock, st):
|
|||
|
|
if sock in active:
|
|||
|
|
try:
|
|||
|
|
sel.unregister(sock)
|
|||
|
|
except (KeyError, ValueError):
|
|||
|
|
pass
|
|||
|
|
active.pop(sock, None)
|
|||
|
|
try:
|
|||
|
|
sock.close()
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
if "verdict" not in st:
|
|||
|
|
_finish(st)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
for ip, port in targets:
|
|||
|
|
st = {"ip": ip, "port": port, "tcp": False, "status": None,
|
|||
|
|
"ctype": "", "body": "", "err": None, "got": b"",
|
|||
|
|
"req": ("GET %s HTTP/1.0\r\nHost: %s\r\n"
|
|||
|
|
"Accept: application/json\r\nConnection: close\r\n\r\n"
|
|||
|
|
% (path, ip)).encode("ascii")}
|
|||
|
|
states.append(st)
|
|||
|
|
sock = None
|
|||
|
|
try:
|
|||
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|||
|
|
sock.setblocking(False)
|
|||
|
|
sock.connect((ip, port)) # 非阻塞：要么立刻 0，要么 BlockingIOError
|
|||
|
|
st["tcp"] = True
|
|||
|
|
except BlockingIOError:
|
|||
|
|
pass # 在途：可写时看 SO_ERROR 才算数
|
|||
|
|
except OSError as e: # 本机立刻就知道连不上（无路由等）
|
|||
|
|
st["tcp"] = False
|
|||
|
|
st["err"] = e.winerror or e.errno or type(e).__name__
|
|||
|
|
if sock is not None:
|
|||
|
|
try:
|
|||
|
|
sock.close()
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
sock = None
|
|||
|
|
_finish(st)
|
|||
|
|
continue
|
|||
|
|
sel.register(sock, selectors.EVENT_WRITE, st)
|
|||
|
|
active[sock] = st
|
|||
|
|
|
|||
|
|
while active:
|
|||
|
|
left = deadline - time.monotonic()
|
|||
|
|
if left <= 0:
|
|||
|
|
break
|
|||
|
|
try:
|
|||
|
|
ready = sel.select(left)
|
|||
|
|
except OSError as e:
|
|||
|
|
errors.append("select:%s" % type(e).__name__)
|
|||
|
|
break
|
|||
|
|
for key, events in ready:
|
|||
|
|
st = key.data
|
|||
|
|
sock = key.fileobj
|
|||
|
|
if st.get("done"):
|
|||
|
|
retire(sock, st)
|
|||
|
|
continue
|
|||
|
|
try:
|
|||
|
|
if events & selectors.EVENT_WRITE and not st.get("sent"):
|
|||
|
|
if sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR):
|
|||
|
|
st["tcp"] = False
|
|||
|
|
st["err"] = "connect-refused"
|
|||
|
|
retire(sock, st)
|
|||
|
|
continue
|
|||
|
|
sock.sendall(st["req"])
|
|||
|
|
st["tcp"] = True # 可写且 SO_ERROR==0：这次真连上了
|
|||
|
|
st["sent"] = True
|
|||
|
|
sel.modify(sock, selectors.EVENT_READ, st)
|
|||
|
|
continue
|
|||
|
|
chunk = sock.recv(4096)
|
|||
|
|
if not chunk:
|
|||
|
|
retire(sock, st) # EOF：对方关连接，够判了
|
|||
|
|
continue
|
|||
|
|
st["got"] += chunk
|
|||
|
|
st["tcp"] = True
|
|||
|
|
# 按 Content-Length 收完就走：路由器多半 keep-alive 不关连接，
|
|||
|
|
# 死等 EOF 会把每个活非板子都拖满截止，整批白跑。
|
|||
|
|
head_end = st["got"].find(b"\r\n\r\n")
|
|||
|
|
if head_end >= 0:
|
|||
|
|
cl = _content_length(st["got"])
|
|||
|
|
have = len(st["got"]) - head_end - 4
|
|||
|
|
if (cl is not None and have >= cl) or have > 65536:
|
|||
|
|
retire(sock, st)
|
|||
|
|
except OSError as e:
|
|||
|
|
if (e.winerror or e.errno) in (10035, 11): # WSAEWOULDBLOCK/EAGAIN
|
|||
|
|
continue
|
|||
|
|
st["err"] = e.winerror or e.errno or type(e).__name__
|
|||
|
|
retire(sock, st)
|
|||
|
|
finally:
|
|||
|
|
for st in states:
|
|||
|
|
if "verdict" not in st:
|
|||
|
|
_finish(st)
|
|||
|
|
for sock in list(active):
|
|||
|
|
try:
|
|||
|
|
sel.unregister(sock)
|
|||
|
|
except (KeyError, ValueError):
|
|||
|
|
pass
|
|||
|
|
try:
|
|||
|
|
sock.close()
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
active.clear()
|
|||
|
|
sel.close()
|
|||
|
|
return states
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _finish(st):
|
|||
|
|
"""把已收字节解析成判决，并把大缓冲释放掉（结果要回给页面，不能揣 64 KB）。"""
|
|||
|
|
status, ctype, body = _parse_http(st.get("got") or b"")
|
|||
|
|
st["status"], st["ctype"] = status, ctype
|
|||
|
|
st["head"] = body[:HEAD_BYTES].replace("\r", "").replace("\n", " ")
|
|||
|
|
if st.get("tcp"):
|
|||
|
|
v = classify(st["ip"], status, ctype, body)
|
|||
|
|
else:
|
|||
|
|
v = {"verdict": "unknown",
|
|||
|
|
"reason": "tcp-failed:%s" % (st.get("err") or "no-connection")}
|
|||
|
|
st.update(v)
|
|||
|
|
st.pop("req", None)
|
|||
|
|
st.pop("got", None)
|
|||
|
|
st["done"] = True
|
|||
|
|
|
|||
|
|
|
|||
|
|
def probe_batch(ips, ports=None, path=PROBE_PATH, timeout_ms=750):
|
|||
|
|
"""逐 IP 探测。返回 {ok, results, scan, errors}；异常只记账，绝不写 stderr。"""
|
|||
|
|
t0 = time.monotonic()
|
|||
|
|
errors = []
|
|||
|
|
scan = {"candidates": 0, "tcpOpen": 0, "httpProbed": 0,
|
|||
|
|
"board": 0, "rejected": 0, "unknown": 0,
|
|||
|
|
"skipped": [], "durationMs": 0, "timeoutMs": 0, "concurrency": CONCURRENCY}
|
|||
|
|
if path not in ALLOWED_PATHS:
|
|||
|
|
errors.append("path-not-allowed:%s" % path)
|
|||
|
|
path = PROBE_PATH
|
|||
|
|
timeout_ms = max(80, min(int(timeout_ms or 750), TIMEOUT_MS_MAX))
|
|||
|
|
scan["timeoutMs"] = timeout_ms
|
|||
|
|
|
|||
|
|
use_ports = [int(p) for p in (ports or [80]) if str(p).strip()]
|
|||
|
|
ips_in = [str(x or "").strip() for x in (ips or [])]
|
|||
|
|
targets = []
|
|||
|
|
seen = set()
|
|||
|
|
# 批大小由服务侧兜死：页面漏分批时也不会让一个 handler 线程在
|
|||
|
|
# 254 个 socket 上挂 24 s。多出来的原样回 pending，由页面下一批接着探。
|
|||
|
|
if len(ips_in) > BATCH_MAX_IPS:
|
|||
|
|
pending_overflow = ["%s:overflow" % ip for ip in ips_in[BATCH_MAX_IPS:]]
|
|||
|
|
errors.append("ips-truncated:%d" % len(pending_overflow))
|
|||
|
|
else:
|
|||
|
|
pending_overflow = []
|
|||
|
|
for ip in ips_in[:BATCH_MAX_IPS]:
|
|||
|
|
if not valid_ipv4(ip):
|
|||
|
|
scan["skipped"].append([ip or "(empty)", "not-ipv4"])
|
|||
|
|
continue
|
|||
|
|
for port in use_ports:
|
|||
|
|
key = (ip, port)
|
|||
|
|
if key in seen:
|
|||
|
|
continue
|
|||
|
|
seen.add(key)
|
|||
|
|
targets.append(key)
|
|||
|
|
scan["candidates"] = len(targets)
|
|||
|
|
scan["requested"] = len(ips_in)
|
|||
|
|
|
|||
|
|
raw = []
|
|||
|
|
pending = list(pending_overflow)
|
|||
|
|
# 一批的总预算 = 分批数 × 单目标截止 + 300ms 余量。页面自己控制批大小(≤24)，
|
|||
|
|
# 所以这里只是兜底：超出预算的尾巴回 pending，让页面继续下一批，绝不静默截断。
|
|||
|
|
n_chunks = (len(targets) + CONCURRENCY - 1) // CONCURRENCY
|
|||
|
|
budget_ms = min(max(timeout_ms * n_chunks + 300, timeout_ms), BUDGET_MS_MAX)
|
|||
|
|
batch_deadline = t0 + budget_ms / 1000.0
|
|||
|
|
scan["budgetMs"] = budget_ms
|
|||
|
|
for i in range(0, len(targets), CONCURRENCY):
|
|||
|
|
chunk = targets[i:i + CONCURRENCY]
|
|||
|
|
now = time.monotonic()
|
|||
|
|
if now >= batch_deadline:
|
|||
|
|
# 批次预算用尽：剩下的一律记账交回页面，绝不伪装成"探过了"。
|
|||
|
|
pending += ["%s:%d" % t for t in chunk]
|
|||
|
|
continue
|
|||
|
|
# 每个 chunk 各有自己的硬截止，但整批再套一个总预算（分批是限流，
|
|||
|
|
# 不是把预算乘 N）。
|
|||
|
|
chunk_deadline = min(batch_deadline, now + timeout_ms / 1000.0)
|
|||
|
|
try:
|
|||
|
|
raw += _probe_chunk(chunk, path, chunk_deadline, errors)
|
|||
|
|
except Exception as e: # noqa: BLE001 - 兜住一切，不外泄
|
|||
|
|
errors.append("chunk:%s:%s" % (type(e).__name__, e))
|
|||
|
|
for ip, port in chunk:
|
|||
|
|
raw.append({"ip": ip, "port": port, "tcp": False, "status": None,
|
|||
|
|
"verdict": "unknown", "reason": "probe-error"})
|
|||
|
|
|
|||
|
|
results = []
|
|||
|
|
for r in raw:
|
|||
|
|
item = {"ip": r.get("ip"), "port": r.get("port"), "tcp": bool(r.get("tcp")),
|
|||
|
|
"verdict": r.get("verdict") or "unknown",
|
|||
|
|
"reason": r.get("reason") or "",
|
|||
|
|
"kind": r.get("kind"), "schema": r.get("schema"),
|
|||
|
|
"mode": r.get("mode"), "connected": r.get("connected"),
|
|||
|
|
"self_ip": r.get("self_ip"), "status": r.get("status"),
|
|||
|
|
"head": r.get("head") or ""}
|
|||
|
|
results.append({k: v for k, v in item.items() if v is not None})
|
|||
|
|
|
|||
|
|
for st in results:
|
|||
|
|
if st.get("tcp"):
|
|||
|
|
scan["tcpOpen"] += 1
|
|||
|
|
if st.get("status") is not None:
|
|||
|
|
scan["httpProbed"] += 1
|
|||
|
|
v = st["verdict"]
|
|||
|
|
scan[v] = scan.get(v, 0) + 1
|
|||
|
|
scan["pending"] = pending
|
|||
|
|
scan["truncated"] = bool(pending)
|
|||
|
|
scan["durationMs"] = int((time.monotonic() - t0) * 1000)
|
|||
|
|
return {"ok": True, "path": path, "results": results,
|
|||
|
|
"scan": scan, "errors": errors}
|