d74ea6cc69
- Replace deprecated dns.servers[] with new format (type + server + server_port) - Add _make_dns_server() to parse udp://, tls://, https:// and bare IP addresses - Drop ENABLE_DEPRECATED_* environment variable shims — fully v1.14-ready - Fix transport block: omit for plain TCP (sing-box default), add httpupgrade support - Add dns.final fallback and route.default_domain_resolver - Config file renamed to hidden /tmp/.sub2socks.json (matches README) - Include test_proxy.py live proxy CONNECT test suite
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Comprehensive SOCKS5 proxy test — run inside sub2socks container."""
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
print("=" * 60)
|
|
print(" Sub2SOCKS Live Proxy Test Suite")
|
|
print("=" * 60)
|
|
|
|
# Test 1: SOCKS5 handshake
|
|
print("\n[1] SOCKS5 Handshake")
|
|
try:
|
|
s = socket.create_connection(("127.0.0.1", 2080), timeout=15)
|
|
s.sendall(b"\x05\x01\x00")
|
|
resp = s.recv(2)
|
|
ok = (resp == b"\x05\x00")
|
|
print(f" Response: {resp.hex():>6} -> {'PASS' if ok else 'FAIL'}")
|
|
except Exception as e:
|
|
print(f" ERROR: {e}")
|
|
ok = None
|
|
|
|
# Test 2: CONNECT requests
|
|
print("\n[2] SOCKS5 CONNECT Targets")
|
|
targets = [
|
|
("api.ipify.org", 443, "ipify API"),
|
|
("httpbin.org", 443, "httpbin "),
|
|
("google.com", 80, "Google HTTP "),
|
|
("cloudflare.com", 443, "Cloudflare "),
|
|
]
|
|
|
|
for host, port, label in targets:
|
|
try:
|
|
s2 = socket.create_connection(("127.0.0.1", 2080), timeout=20)
|
|
s2.sendall(b"\x05\x01\x00")
|
|
s2.recv(2)
|
|
|
|
hbytes = host.encode()
|
|
req = (b"\x05\x01\x00\x03"
|
|
+ bytes([len(hbytes)])
|
|
+ hbytes
|
|
+ struct.pack("!H", port))
|
|
|
|
t0 = time.monotonic()
|
|
s2.sendall(req)
|
|
conn_resp = s2.recv(8)
|
|
elapsed = (time.monotonic() - t0) * 1000
|
|
status_byte = conn_resp[1]
|
|
if status_byte == 0:
|
|
print(f" {label:13s} :{port:<4} SUCCESS ({elapsed:.0f} ms)")
|
|
else:
|
|
print(f" {label:13s} :{port:<4} FAIL (code={status_byte})")
|
|
s2.close()
|
|
except socket.timeout:
|
|
print(f" {label:13s} :{port:<4} TIMEOUT")
|
|
except Exception as e:
|
|
print(f" {label:13s} :{port:<4} ERROR: {e}")
|
|
|
|
# Test 3: Process health
|
|
print("\n[3] Process health")
|
|
try:
|
|
with open("/proc/1/cmdline", "rb") as f:
|
|
cmdline = f.read().decode("utf-8", errors="replace").replace("\x00", " ")
|
|
has_sb = "sing-box" in cmdline
|
|
print(f" PID 1: {'sing-box OK' if has_sb else cmdline[:80]}")
|
|
except Exception as e:
|
|
print(f" ERROR reading /proc/1: {e}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print(" Tests complete")
|
|
print("=" * 60)
|