102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
EO 优选 IP 扫描 v2 (schannel 证书校验版)
|
|
判据更严格: 用 curl (Windows schannel, 系统证书链) + --resolve 强制 SNI,
|
|
只有当 TLS 证书链校验通过(浏览器视角可访问)且返回 HTTP 响应才算 hit。
|
|
这能区分 "CDN 通用池 IP (证书 *.cdn.myqcloud.com 不匹配)" 与 "真正承载目标站点的 EO 边缘"。
|
|
用法: python eo-probe-v2.py --cidrs 43.174.246.0/24 43.174.247.0/24 ...
|
|
"""
|
|
import argparse
|
|
import concurrent.futures
|
|
import json
|
|
import subprocess
|
|
import time
|
|
|
|
DEFAULT_CIDRS = [
|
|
"43.174.150.0/24", "43.174.151.0/24", "43.174.152.0/24", "43.174.153.0/24",
|
|
"43.174.78.0/24", "43.174.79.0/24", "43.174.57.0/24",
|
|
"43.175.130.0/24", "43.175.131.0/24", "43.175.132.0/24", "43.175.133.0/24",
|
|
"43.175.150.0/24", "43.175.151.0/24", "43.175.152.0/24",
|
|
"43.175.165.0/24", "43.175.166.0/24",
|
|
"43.175.185.0/24", "43.175.186.0/24",
|
|
"43.174.246.0/24", "43.174.247.0/24", # EO 官方当前分配段(新加坡)
|
|
]
|
|
|
|
def cidr_hosts(cidr):
|
|
net, bits = cidr.split("/")
|
|
bits = int(bits)
|
|
a, b, c, d = map(int, net.split("."))
|
|
base = (a << 24) + (b << 16) + (c << 8) + d
|
|
count = 1 << (32 - bits)
|
|
for i in range(1, count - 1):
|
|
v = base + i
|
|
yield f"{(v >> 24) & 255}.{(v >> 16) & 255}.{(v >> 8) & 255}.{v & 255}"
|
|
|
|
def probe(ip, host, timeout):
|
|
t0 = time.perf_counter()
|
|
# schannel on Windows: 默认走系统 CA, 证书不匹配会直接失败 => http_code 000 + 错误
|
|
cmd = ["curl", "-sS", "--resolve", f"{host}:443:{ip}",
|
|
"--connect-timeout", str(timeout), "--max-time", str(timeout * 2),
|
|
"-o", "NUL", "-w", "%{http_code}|%{time_total}|%{time_starttransfer}",
|
|
f"https://{host}/"]
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout * 3,
|
|
errors="replace")
|
|
out = r.stdout.strip()
|
|
parts = out.split("|")
|
|
code = parts[0] if parts else "000"
|
|
el = (time.perf_counter() - t0) * 1000
|
|
err = r.stderr.strip() if r.stderr else ""
|
|
if code not in ("000",):
|
|
return {"ip": ip, "http": code, "total_s": float(parts[1]) if len(parts) > 1 else None,
|
|
"elapsed_ms": round(el, 1), "note": err.splitlines()[-1] if err else ""}
|
|
# 000: 看看是不是证书错误
|
|
note = ""
|
|
if "SEC_E_WRONG_PRINCIPAL" in err or "certificate" in err.lower():
|
|
note = "cert-mismatch"
|
|
elif err:
|
|
note = err.splitlines()[-1][:80]
|
|
return None
|
|
except Exception as e:
|
|
return None
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--host", default="box.shiroko.one")
|
|
ap.add_argument("--cidrs", nargs="*", default=None)
|
|
ap.add_argument("--workers", type=int, default=64)
|
|
ap.add_argument("--timeout", type=float, default=3)
|
|
ap.add_argument("--out", default="eo-hits-v2.json")
|
|
args = ap.parse_args()
|
|
|
|
cidrs = args.cidrs or DEFAULT_CIDRS
|
|
ips = []
|
|
for c in cidrs:
|
|
ips.extend(cidr_hosts(c))
|
|
print(f"[*] host={args.host} 段={len(cidrs)} IP={len(ips)} workers={args.workers}")
|
|
|
|
hits = []
|
|
t_start = time.perf_counter()
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as ex:
|
|
futs = {ex.submit(probe, ip, args.host, args.timeout): ip for ip in ips}
|
|
done = 0
|
|
for fut in concurrent.futures.as_completed(futs):
|
|
done += 1
|
|
r = fut.result()
|
|
if r:
|
|
hits.append(r)
|
|
print(f" HIT {r['ip']:16s} http={r['http']} total={r['total_s']}s {r['note']}")
|
|
if done % 500 == 0:
|
|
print(f" [{done}/{len(ips)}] hits={len(hits)} elapsed={time.perf_counter()-t_start:.0f}s")
|
|
|
|
print(f"\n[*] done in {time.perf_counter()-t_start:.0f}s, hits={len(hits)}")
|
|
if hits:
|
|
hits.sort(key=lambda x: x["total_s"] or 99)
|
|
with open(args.out, "w", encoding="utf-8") as f:
|
|
json.dump(hits, f, ensure_ascii=False, indent=2)
|
|
print(f"[*] saved: {args.out}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|