也可以,我附在这里,你看看。 或者你可以试试让Codex写一个
- #!/usr/bin/env python3
- """Probe SOCKS5 UDP and Vodafone WiFi Calling IKE reachability."""
- from __future__ import annotations
- import argparse
- import hashlib
- import json
- import os
- import secrets
- import socket
- import struct
- import sys
- import time
- from dataclasses import asdict, dataclass
- from datetime import datetime, timezone
- from typing import Callable, Iterable
- DEFAULT_EPDG_HOST = "epdg.epc.mnc010.mcc234.pub.3gppnetwork.org"
- DEFAULT_DNS_TARGETS = (("dns_cloudflare", "1.1.1.1"), ("dns_google", "8.8.8.8"))
- # RFC 3526 MODP group 14.
- MODP_2048_PRIME = int(
- "FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1"
- "29024E088A67CC74020BBEA63B139B22514A08798E3404DD"
- "EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245"
- "E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED"
- "EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D"
- "C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F"
- "83655D23DCA3AD961C62F356208552BB9ED529077096966D"
- "670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B"
- "E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9"
- "DE2BCBF6955817183995497CEA956AE515D2261898FA0510"
- "15728E5A8AACAA68FFFFFFFFFFFFFFFF",
- 16,
- )
- @dataclass
- class ProbeResult:
- name: str
- target: str
- passed: bool
- status: str
- rtt_ms: float | None = None
- bytes_received: int = 0
- detail: str = ""
- def parse_proxy(value: str) -> tuple[str, int]:
- host, separator, raw_port = value.rpartition(":")
- if not separator or not host:
- raise argparse.ArgumentTypeError("proxy must use HOST:PORT")
- try:
- port = int(raw_port)
- except ValueError as error:
- raise argparse.ArgumentTypeError("proxy port must be an integer") from error
- if not 1 <= port <= 65535:
- raise argparse.ArgumentTypeError("proxy port must be between 1 and 65535")
- return host, port
- def recv_exact(sock: socket.socket, size: int) -> bytes:
- chunks = bytearray()
- while len(chunks) < size:
- chunk = sock.recv(size - len(chunks))
- if not chunk:
- raise RuntimeError("SOCKS5 control connection closed")
- chunks.extend(chunk)
- return bytes(chunks)
- def encode_socks_udp(ip: str, port: int, payload: bytes) -> bytes:
- return b"\x00\x00\x00\x01" + socket.inet_aton(ip) + struct.pack("!H", port) + payload
- def decode_socks_udp(frame: bytes) -> tuple[str, int, bytes]:
- if len(frame) < 10 or frame[:3] != b"\x00\x00\x00":
- raise ValueError("invalid SOCKS5 UDP frame")
- address_type = frame[3]
- offset = 4
- if address_type == 1:
- host = socket.inet_ntoa(frame[offset : offset + 4])
- offset += 4
- elif address_type == 3:
- name_length = frame[offset]
- offset += 1
- host = frame[offset : offset + name_length].decode("ascii")
- offset += name_length
- elif address_type == 4:
- host = socket.inet_ntop(socket.AF_INET6, frame[offset : offset + 16])
- offset += 16
- else:
- raise ValueError(f"unsupported SOCKS5 address type: {address_type}")
- if len(frame) < offset + 2:
- raise ValueError("truncated SOCKS5 UDP frame")
- port = struct.unpack("!H", frame[offset : offset + 2])[0]
- return host, port, frame[offset + 2 :]
- class Socks5UdpClient:
- def __init__(self, proxy: tuple[str, int], timeout: float):
- self.proxy = proxy
- self.timeout = timeout
- self.control: socket.socket | None = None
- self.udp: socket.socket | None = None
- self.relay: tuple[str, int] | None = None
- def __enter__(self) -> "Socks5UdpClient":
- self.udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- self.udp.bind(("127.0.0.1", 0))
- local_host, local_port = self.udp.getsockname()
- self.control = socket.create_connection(self.proxy, timeout=self.timeout)
- self.control.settimeout(self.timeout)
- self.control.sendall(b"\x05\x01\x00")
- if recv_exact(self.control, 2) != b"\x05\x00":
- raise RuntimeError("SOCKS5 proxy requires unsupported authentication")
- request = (
- b"\x05\x03\x00\x01"
- + socket.inet_aton(local_host)
- + struct.pack("!H", local_port)
- )
- self.control.sendall(request)
- header = recv_exact(self.control, 4)
- if header[:2] != b"\x05\x00":
- raise RuntimeError(f"SOCKS5 UDP ASSOCIATE failed with code {header[1]}")
- relay_host = self._read_address(header[3])
- relay_port = struct.unpack("!H", recv_exact(self.control, 2))[0]
- if relay_host in ("0.0.0.0", "::"):
- relay_host = self.proxy[0]
- self.relay = (relay_host, relay_port)
- return self
- def _read_address(self, address_type: int) -> str:
- assert self.control is not None
- if address_type == 1:
- return socket.inet_ntoa(recv_exact(self.control, 4))
- if address_type == 3:
- length = recv_exact(self.control, 1)[0]
- return recv_exact(self.control, length).decode("ascii")
- if address_type == 4:
- return socket.inet_ntop(socket.AF_INET6, recv_exact(self.control, 16))
- raise RuntimeError(f"unsupported SOCKS5 relay address type: {address_type}")
- def request(
- self,
- target_ip: str,
- target_port: int,
- payload: bytes,
- validator: Callable[[bytes], bool],
- ) -> ProbeResult:
- assert self.udp is not None and self.relay is not None
- self.udp.settimeout(self.timeout)
- started = time.monotonic()
- self.udp.sendto(encode_socks_udp(target_ip, target_port, payload), self.relay)
- deadline = started + self.timeout
- while True:
- remaining = deadline - time.monotonic()
- if remaining <= 0:
- return ProbeResult(
- name="udp",
- target=f"{target_ip}:{target_port}",
- passed=False,
- status="timeout",
- )
- self.udp.settimeout(remaining)
- try:
- frame, _ = self.udp.recvfrom(65535)
- except socket.timeout:
- return ProbeResult(
- name="udp",
- target=f"{target_ip}:{target_port}",
- passed=False,
- status="timeout",
- )
- try:
- _, _, response = decode_socks_udp(frame)
- except ValueError:
- continue
- if not validator(response):
- continue
- return ProbeResult(
- name="udp",
- target=f"{target_ip}:{target_port}",
- passed=True,
- status="reply",
- rtt_ms=round((time.monotonic() - started) * 1000, 1),
- bytes_received=len(response),
- )
- def request_many_ike(
- self, targets: Iterable[str], port: int
- ) -> list[ProbeResult]:
- assert self.udp is not None and self.relay is not None
- pending: dict[bytes, tuple[str, float]] = {}
- marker = port == 4500
- source_port = self.udp.getsockname()[1]
- for target_ip in targets:
- initiator_spi, packet = build_ikev2_request(
- target_ip, port, marker, source_port
- )
- pending[initiator_spi] = (target_ip, time.monotonic())
- self.udp.sendto(encode_socks_udp(target_ip, port, packet), self.relay)
- results: dict[str, ProbeResult] = {}
- deadline = time.monotonic() + self.timeout
- while pending and time.monotonic() < deadline:
- self.udp.settimeout(max(0.1, deadline - time.monotonic()))
- try:
- frame, _ = self.udp.recvfrom(65535)
- except socket.timeout:
- break
- try:
- _, _, response = decode_socks_udp(frame)
- except ValueError:
- continue
- ike = response[4:] if response.startswith(b"\x00\x00\x00\x00") else response
- parsed = parse_ikev2_response(ike)
- if parsed is None:
- continue
- initiator_spi, valid = parsed
- if initiator_spi not in pending:
- continue
- target_ip, started = pending.pop(initiator_spi)
- results[target_ip] = ProbeResult(
- name=f"ikev2_udp{port}",
- target=f"{target_ip}:{port}",
- passed=valid,
- status="valid_ikev2_reply" if valid else "invalid_ikev2_reply",
- rtt_ms=round((time.monotonic() - started) * 1000, 1),
- bytes_received=len(ike),
- )
- for target_ip, _ in pending.values():
- results[target_ip] = ProbeResult(
- name=f"ikev2_udp{port}",
- target=f"{target_ip}:{port}",
- passed=False,
- status="timeout",
- )
- return [results[target_ip] for target_ip in targets]
- def __exit__(self, *_: object) -> None:
- if self.udp is not None:
- self.udp.close()
- if self.control is not None:
- self.control.close()
- def socks5_tcp_connect(proxy: tuple[str, int], timeout: float) -> ProbeResult:
- started = time.monotonic()
- try:
- with socket.create_connection(proxy, timeout=timeout) as control:
- control.settimeout(timeout)
- control.sendall(b"\x05\x01\x00")
- if recv_exact(control, 2) != b"\x05\x00":
- raise RuntimeError("authentication required")
- host = b"www.cloudflare.com"
- request = b"\x05\x01\x00\x03" + bytes([len(host)]) + host + struct.pack("!H", 443)
- control.sendall(request)
- header = recv_exact(control, 4)
- if header[:2] != b"\x05\x00":
- raise RuntimeError(f"CONNECT failed with code {header[1]}")
- address_type = header[3]
- if address_type == 1:
- recv_exact(control, 4)
- elif address_type == 3:
- recv_exact(control, recv_exact(control, 1)[0])
- elif address_type == 4:
- recv_exact(control, 16)
- else:
- raise RuntimeError("invalid CONNECT reply")
- recv_exact(control, 2)
- return ProbeResult(
- name="socks5_tcp_connect",
- target="www.cloudflare.com:443",
- passed=True,
- status="connected",
- rtt_ms=round((time.monotonic() - started) * 1000, 1),
- )
- except (OSError, RuntimeError) as error:
- return ProbeResult(
- name="socks5_tcp_connect",
- target="www.cloudflare.com:443",
- passed=False,
- status="failed",
- detail=str(error),
- )
- def build_dns_query(name: str = "example.com") -> tuple[bytes, Callable[[bytes], bool]]:
- transaction_id = os.urandom(2)
- qname = b"".join(bytes([len(label)]) + label.encode("ascii") for label in name.split(".")) + b"\x00"
- packet = (
- transaction_id
- + struct.pack("!HHHHH", 0x0100, 1, 0, 0, 0)
- + qname
- + struct.pack("!HH", 1, 1)
- )
- def validate(response: bytes) -> bool:
- return (
- len(response) >= 12
- and response[:2] == transaction_id
- and bool(response[2] & 0x80)
- )
- return packet, validate
- def build_stun_query() -> tuple[bytes, Callable[[bytes], bool]]:
- transaction_id = os.urandom(12)
- packet = struct.pack("!HHI12s", 0x0001, 0, 0x2112A442, transaction_id)
- def validate(response: bytes) -> bool:
- return (
- len(response) >= 20
- and response[:2] == b"\x01\x01"
- and response[4:8] == b"\x21\x12\xa4\x42"
- and response[8:20] == transaction_id
- )
- return packet, validate
- def ike_transform(
- more: bool, transform_type: int, transform_id: int, attributes: bytes = b""
- ) -> bytes:
- return struct.pack(
- "!BBHBBH",
- 3 if more else 0,
- 0,
- 8 + len(attributes),
- transform_type,
- 0,
- transform_id,
- ) + attributes
- def ike_notify(next_payload: int, notify_type: int, data: bytes = b"") -> bytes:
- body = struct.pack("!BBH", 0, 0, notify_type) + data
- return struct.pack("!BBH", next_payload, 0, 4 + len(body)) + body
- def nat_detection_hash(initiator_spi: bytes, ip: str, port: int) -> bytes:
- return hashlib.sha1(
- initiator_spi
- + b"\x00" * 8
- + socket.inet_aton(ip)
- + struct.pack("!H", port)
- ).digest()
- def build_ikev2_request(
- target_ip: str,
- target_port: int,
- marker: bool,
- source_port: int = 45000,
- ) -> tuple[bytes, bytes]:
- initiator_spi = os.urandom(8)
- transforms = b"".join(
- (
- ike_transform(True, 1, 12, struct.pack("!HH", 0x800E, 256)),
- ike_transform(True, 2, 5),
- ike_transform(True, 3, 12),
- ike_transform(False, 4, 14),
- )
- )
- proposal = struct.pack(
- "!BBHBBBB", 0, 0, 8 + len(transforms), 1, 1, 0, 4
- ) + transforms
- security_association = struct.pack("!BBH", 34, 0, 4 + len(proposal)) + proposal
- public_value = pow(
- 2, secrets.randbits(320) | 1, MODP_2048_PRIME
- ).to_bytes(256, "big")
- key_exchange = struct.pack("!BBHHH", 40, 0, 264, 14, 0) + public_value
- nonce_data = os.urandom(32)
- nonce = struct.pack("!BBH", 41, 0, 4 + len(nonce_data)) + nonce_data
- source_nat = ike_notify(
- 41, 16388, nat_detection_hash(initiator_spi, "127.0.0.1", source_port)
- )
- destination_nat = ike_notify(
- 41, 16389, nat_detection_hash(initiator_spi, target_ip, target_port)
- )
- fragmentation = ike_notify(0, 16430)
- body = security_association + key_exchange + nonce + source_nat + destination_nat + fragmentation
- header = struct.pack(
- "!8s8sBBBBII",
- initiator_spi,
- b"\x00" * 8,
- 33,
- 0x20,
- 34,
- 0x08,
- 0,
- 28 + len(body),
- )
- packet = header + body
- return initiator_spi, (b"\x00\x00\x00\x00" + packet if marker else packet)
- def parse_ikev2_response(packet: bytes) -> tuple[bytes, bool] | None:
- if len(packet) < 28:
- return None
- fields = struct.unpack("!8s8sBBBBII", packet[:28])
- initiator_spi, responder_spi, _, version, exchange, flags, _, length = fields
- valid = (
- responder_spi != b"\x00" * 8
- and version >> 4 == 2
- and exchange == 34
- and bool(flags & 0x20)
- and 28 <= length <= len(packet)
- )
- return initiator_spi, bool(valid)
- def resolve_ipv4(host: str) -> list[str]:
- return sorted(
- {
- item[4][0]
- for item in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_DGRAM)
- }
- )
- def classify(results: list[ProbeResult]) -> str:
- ike500 = [result for result in results if result.name == "ikev2_udp500"]
- ike4500 = [result for result in results if result.name == "ikev2_udp4500"]
- if not any(result.passed for result in ike500) or not any(
- result.passed for result in ike4500
- ):
- return "FAIL"
- if not all(result.passed for result in ike500 + ike4500):
- return "PARTIAL"
- return "PASS"
- def run(args: argparse.Namespace) -> tuple[list[ProbeResult], list[str]]:
- results = [socks5_tcp_connect(args.proxy, args.timeout)]
- resolved_epdg_ips = args.epdg_ip or resolve_ipv4(args.epdg_host)
- epdg_ips = (
- resolved_epdg_ips
- if args.epdg_ip or args.all_epdg
- else resolved_epdg_ips[:1]
- )
- # Keep control traffic and each IKE port in separate UDP associations.
- # Some proxy chains retain destination-specific UDP state after a timeout.
- with Socks5UdpClient(args.proxy, args.timeout) as client:
- for name, target_ip in DEFAULT_DNS_TARGETS:
- packet, validator = build_dns_query()
- result = client.request(target_ip, 53, packet, validator)
- result.name = name
- results.append(result)
- stun_ip = socket.gethostbyname("stun.l.google.com")
- packet, validator = build_stun_query()
- result = client.request(stun_ip, 19302, packet, validator)
- result.name = "stun_google"
- result.detail = "informational; not a WiFi Calling gate"
- results.append(result)
- with Socks5UdpClient(args.proxy, args.timeout) as client:
- results.extend(client.request_many_ike(epdg_ips, 500))
- with Socks5UdpClient(args.proxy, args.timeout) as client:
- results.extend(client.request_many_ike(epdg_ips, 4500))
- return results, epdg_ips
- def format_text(
- args: argparse.Namespace, results: list[ProbeResult], epdg_ips: list[str]
- ) -> str:
- lines = [
- f"Proxy: {args.proxy[0]}:{args.proxy[1]}",
- f"Node: {args.node_label or '(not specified)'}",
- f"ePDG: {args.epdg_host} -> {', '.join(epdg_ips)}",
- "",
- ]
- for result in results:
- state = "PASS" if result.passed else "FAIL"
- timing = f" {result.rtt_ms:.1f}ms" if result.rtt_ms is not None else ""
- lines.append(
- f"[{state}] {result.name:<20} {result.target:<24} {result.status}{timing}"
- )
- verdict = classify(results)
- lines.extend(
- (
- "",
- f"Verdict: {verdict}",
- "PASS proves pre-auth IKE reachability, not carrier account/SIM authentication.",
- )
- )
- return "\n".join(lines)
- def format_json(
- args: argparse.Namespace, results: list[ProbeResult], epdg_ips: list[str]
- ) -> str:
- payload = {
- "timestamp_utc": datetime.now(timezone.utc).isoformat(),
- "proxy": f"{args.proxy[0]}:{args.proxy[1]}",
- "node_label": args.node_label or None,
- "epdg_host": args.epdg_host,
- "epdg_ips": epdg_ips,
- "verdict": classify(results),
- "results": [asdict(result) for result in results],
- "caveat": "IKE probes verify pre-auth reachability, not carrier account/SIM authentication.",
- }
- return json.dumps(payload, ensure_ascii=False, indent=2)
- def format_ticket(
- args: argparse.Namespace, results: list[ProbeResult], epdg_ips: list[str]
- ) -> str:
- verdict = classify(results)
- failed = [result for result in results if not result.passed]
- passed = [result for result in results if result.passed]
- lines = [
- "Subject: UDP relay / WiFi Calling IKE reachability issue",
- "",
- f"Test time (UTC): {datetime.now(timezone.utc).isoformat()}",
- f"Node: {args.node_label or '[fill in node name]'}",
- f"Local proxy: SOCKS5 {args.proxy[0]}:{args.proxy[1]}",
- f"Target: {args.epdg_host} ({', '.join(epdg_ips)})",
- f"Automated verdict: {verdict}",
- "",
- "Expected:",
- "The node should relay UDP request/response traffic, including IKEv2 UDP 500 and NAT-T UDP 4500.",
- "",
- "Passed checks:",
- ]
- lines.extend(
- f"- {result.name} {result.target}: {result.status}"
- for result in passed
- )
- lines.append("")
- lines.append("Failed checks:")
- lines.extend(
- f"- {result.name} {result.target}: {result.status}"
- for result in failed
- )
- if not failed:
- lines.append("- None")
- lines.extend(
- (
- "",
- "Request:",
- "Please verify server-side UDP relay, firewall rules, NAT mappings, and outbound UDP 500/4500 for this node.",
- "",
- "Note: STUN is informational. IKE results are valid protocol responses but do not test carrier account/SIM authentication.",
- )
- )
- return "\n".join(lines)
- def build_parser() -> argparse.ArgumentParser:
- parser = argparse.ArgumentParser(
- description="Test a local SOCKS5 node for UDP DNS, STUN, and WiFi Calling IKE reachability."
- )
- parser.add_argument(
- "--proxy",
- required=True,
- type=parse_proxy,
- metavar="HOST:PORT",
- help="local no-auth SOCKS5 or Mihomo mixed listener",
- )
- parser.add_argument("--node-label", default="", help="node name included in reports")
- parser.add_argument("--epdg-host", default=DEFAULT_EPDG_HOST)
- parser.add_argument(
- "--epdg-ip",
- action="append",
- default=[],
- help="override resolved ePDG IPv4 address; repeat for multiple addresses",
- )
- parser.add_argument(
- "--all-epdg",
- action="store_true",
- help="probe every resolved ePDG address instead of one address",
- )
- parser.add_argument("--timeout", type=float, default=4.0, help="timeout per probe in seconds")
- parser.add_argument(
- "--format", choices=("text", "json", "ticket"), default="text"
- )
- return parser
- def main() -> int:
- args = build_parser().parse_args()
- try:
- results, epdg_ips = run(args)
- except (OSError, RuntimeError) as error:
- print(f"Probe setup failed: {error}", file=sys.stderr)
- return 1
- if not epdg_ips:
- print("No ePDG IPv4 addresses resolved", file=sys.stderr)
- return 1
- if args.format == "json":
- print(format_json(args, results, epdg_ips))
- elif args.format == "ticket":
- print(format_ticket(args, results, epdg_ips))
- else:
- print(format_text(args, results, epdg_ips))
- verdict = classify(results)
- return 0 if verdict == "PASS" else 2 if verdict == "PARTIAL" else 1
- if __name__ == "__main__":
- raise SystemExit(main())
复制代码 |