lou
af43b07dc0
- client.py: 删未使用导入 TYPE_AUTH_OK/TYPE_DNS_RESP/TYPE_HTTP_RESP - dnsmsg.py: flags 解包后未用 -> 补 QR=1 响应拒绝(符合'非 query 返回 None'语义) - httpmsg.py: headers default_factory=list 推断 Unknown -> lambda: [] 显式类型 - smoke_channel.py: 删未使用导入 json/os/socket/FrameReader/pack_frame; cfg 显式 dict[str, object] - analyze_capture.py: proto Counter() 无类型参数 -> Counter[int]
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""端到端冒烟: 本机起 CloudServer + ChannelClient, 验证 AUTH/帧/HTTP/DNS 链路。
|
|
|
|
用法: .venv/bin/python scripts/smoke_channel.py
|
|
- 生成临时自签证书(/tmp/smoke-tls/)
|
|
- 本机起 HTTP 回显目标(18080) + CloudServer(18443, forward_scheme=http)
|
|
- ChannelClient 连 127.0.0.1:18443, 验证: AUTH、HTTP 套壳转发、DNS 帧往返
|
|
DNS 目标走公网 223.5.5.5, 学校网络不通时只验证帧链路(ERROR/超时也算通路)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import http.server
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
from channel.client import ChannelClient
|
|
from channel.httpmsg import parse_request
|
|
from channel.server import CloudConfig, CloudServer
|
|
|
|
WORK = Path(tempfile.mkdtemp(prefix="smoke-channel-"))
|
|
CERT = WORK / "cert.pem"
|
|
KEY = WORK / "key.pem"
|
|
TARGET_PORT = 18080
|
|
CLOUD_PORT = 18443
|
|
TOKEN = "smoke-token"
|
|
|
|
|
|
def _make_cert() -> None:
|
|
subprocess.run(
|
|
[
|
|
"openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
|
|
"-keyout", str(KEY), "-out", str(CERT), "-days", "1",
|
|
"-subj", "/CN=127.0.0.1",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
|
|
|
|
class _Echo(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
body = ("echo:" + self.path).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
pass
|
|
|
|
|
|
def _run_target() -> None:
|
|
srv = http.server.ThreadingHTTPServer(("127.0.0.1", TARGET_PORT), _Echo)
|
|
srv.serve_forever()
|
|
|
|
|
|
def main() -> int:
|
|
_make_cert()
|
|
target = threading.Thread(target=_run_target, daemon=True)
|
|
target.start()
|
|
|
|
cfg: dict[str, object] = {
|
|
"cloud_side": {
|
|
"listen_port": CLOUD_PORT,
|
|
"token": TOKEN,
|
|
"dns_upstream": "223.5.5.5",
|
|
"forward_scheme": "http",
|
|
"ua": "SmokeUA/1.0",
|
|
"deny_suffixes": [],
|
|
"tls": {"cert_file": str(CERT), "key_file": str(KEY)},
|
|
}
|
|
}
|
|
conf = CloudConfig(cfg)
|
|
srv = CloudServer(("127.0.0.1", CLOUD_PORT), conf)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
|
|
client = ChannelClient(
|
|
host="127.0.0.1", port=CLOUD_PORT, token=TOKEN, retries=1
|
|
)
|
|
client.connect()
|
|
print("[smoke] AUTH ok")
|
|
|
|
raw_req = (
|
|
f"GET /ping HTTP/1.1\r\nHost: 127.0.0.1:{TARGET_PORT}\r\n"
|
|
"User-Agent: curl/8\r\n\r\n"
|
|
).encode()
|
|
resp = client.send_http_request(raw_req)
|
|
assert resp is not None, "http resp is None"
|
|
assert resp.startswith(b"HTTP/1.1 200"), resp[:60]
|
|
parsed = parse_request(raw_req)
|
|
assert parsed is not None
|
|
assert b"echo:/ping" in resp
|
|
print(f"[smoke] HTTP 套壳转发 ok ({len(resp)}B 响应)")
|
|
|
|
qname = b"www.baidu.com"
|
|
query = (
|
|
b"\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00"
|
|
+ b"\x03www\x05baidu\x03com\x00\x00\x01\x00\x01"
|
|
)
|
|
dns_resp = client.send_dns_query(query)
|
|
if dns_resp is not None and len(dns_resp) >= 12:
|
|
print(f"[smoke] DNS 帧往返 ok ({len(dns_resp)}B, qname={qname.decode()})")
|
|
else:
|
|
print("[smoke] DNS 帧往返: ERROR/超时(学校网络公网 DNS 不通属正常), 链路本身已通")
|
|
|
|
client.close()
|
|
srv.shutdown()
|
|
print("[smoke] PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|