100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
|
"""proxy_dns 降级路径实测(模拟云通道挂掉)。
|
||
|
|
|
||
|
|
验证两条降级链:
|
||
|
|
1. 通道失败 -> 直连 fallback_dns 解析成功(全屋 DNS 不断)
|
||
|
|
2. 通道失败 + 直连也失败 -> 回 SERVFAIL(客户端立即切备用 DNS)
|
||
|
|
|
||
|
|
用法: python3 scripts/test_dns_fallback.py
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import socket
|
||
|
|
import sys
|
||
|
|
import threading
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
|
|
|
||
|
|
from channel.client import ChannelClient, ChannelError # noqa: E402
|
||
|
|
from channel.proxy_dns import DnsSplitProxy # noqa: E402
|
||
|
|
|
||
|
|
QUERY = bytes.fromhex(
|
||
|
|
"abcd01000001000000000000037777770674616f62616f03636f6d0000010001"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class DownClient(ChannelClient):
|
||
|
|
"""模拟云通道不可用(所有请求抛 ChannelError)。"""
|
||
|
|
|
||
|
|
def __init__(self) -> None:
|
||
|
|
super().__init__(host="127.0.0.1", port=1, token="test-stub")
|
||
|
|
|
||
|
|
def send_dns_query(self, raw_query: bytes) -> bytes | None:
|
||
|
|
raise ChannelError("simulated channel down")
|
||
|
|
|
||
|
|
def ensure_connected(self) -> None:
|
||
|
|
raise ChannelError("simulated channel down")
|
||
|
|
|
||
|
|
|
||
|
|
def ask(port: int, timeout: float = 6.0) -> bytes:
|
||
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
|
|
s.settimeout(timeout)
|
||
|
|
try:
|
||
|
|
s.sendto(QUERY, ("127.0.0.1", port))
|
||
|
|
resp, _ = s.recvfrom(4096)
|
||
|
|
return resp
|
||
|
|
finally:
|
||
|
|
s.close()
|
||
|
|
|
||
|
|
|
||
|
|
def rcode(resp: bytes) -> int:
|
||
|
|
return resp[3] & 0x0F if len(resp) >= 12 else -1
|
||
|
|
|
||
|
|
|
||
|
|
def case_fallback_ok() -> bool:
|
||
|
|
"""通道挂 + 直连上游可用 -> 应返回正常解析(RCODE=0)。"""
|
||
|
|
proxy = DnsSplitProxy(
|
||
|
|
DownClient(), listen_port=18053, school_dns="223.5.5.5",
|
||
|
|
internal_suffixes=[], fallback_dns="223.5.5.5",
|
||
|
|
)
|
||
|
|
threading.Thread(target=proxy.serve, daemon=True).start()
|
||
|
|
time.sleep(0.3)
|
||
|
|
resp = ask(18053)
|
||
|
|
ok = len(resp) > 12 and rcode(resp) == 0 and proxy.fallback_resolved >= 1
|
||
|
|
print(
|
||
|
|
f" 用例1 通道挂->直连降级: 响应 {len(resp)}B RCODE={rcode(resp)} "
|
||
|
|
f"fallback_resolved={proxy.fallback_resolved} -> "
|
||
|
|
f"{'PASS' if ok else 'FAIL'}"
|
||
|
|
)
|
||
|
|
return ok
|
||
|
|
|
||
|
|
|
||
|
|
def case_servfail() -> bool:
|
||
|
|
"""通道挂 + 直连上游也不可达 -> 应回 SERVFAIL(RCODE=2), 不静默丢弃。"""
|
||
|
|
proxy = DnsSplitProxy(
|
||
|
|
DownClient(), listen_port=18054, school_dns="192.0.2.1",
|
||
|
|
internal_suffixes=[], fallback_dns="192.0.2.1",
|
||
|
|
)
|
||
|
|
threading.Thread(target=proxy.serve, daemon=True).start()
|
||
|
|
time.sleep(0.3)
|
||
|
|
resp = ask(18054)
|
||
|
|
ok = len(resp) >= 12 and rcode(resp) == 2
|
||
|
|
print(
|
||
|
|
f" 用例2 通道+直连全挂->SERVFAIL: 响应 {len(resp)}B "
|
||
|
|
f"RCODE={rcode(resp)} -> {'PASS' if ok else 'FAIL'}"
|
||
|
|
)
|
||
|
|
return ok
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
print("[DNS 降级测试] 模拟云通道不可用")
|
||
|
|
r1 = case_fallback_ok()
|
||
|
|
r2 = case_servfail()
|
||
|
|
print(f"结果: {'全部 PASS' if r1 and r2 else '有 FAIL'}")
|
||
|
|
return 0 if (r1 and r2) else 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|