190 lines
9.6 KiB
Python
190 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Real TCP/UDP/VLESS/TUN and lifecycle checks. Run only inside an isolated user+network namespace.
|
|
|
|
cargo build --manifest-path src-tauri/Cargo.toml --no-default-features --example test_driver
|
|
unshare --user --map-root-user --net python3 tests/netns_integration.py
|
|
|
|
No Internet, real credentials, host routes, sudo, or persistent capabilities required.
|
|
"""
|
|
import http.server
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import select
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
CORE = ROOT / "binaries/sing-box"
|
|
DRIVER = ROOT / "src-tauri/target/debug/examples/test_driver"
|
|
SERVER_IP = "192.0.2.2"
|
|
|
|
def run(*args):
|
|
return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT).strip()
|
|
|
|
def link_names():
|
|
return {link["ifname"] for link in json.loads(run("ip", "-j", "link", "show"))}
|
|
|
|
def wait_for(predicate, seconds=8):
|
|
until = time.monotonic() + seconds
|
|
while time.monotonic() < until:
|
|
if predicate():
|
|
return
|
|
time.sleep(.05)
|
|
raise AssertionError("Timed out waiting for cleanup/readiness")
|
|
|
|
def serve(directory):
|
|
directory = Path(directory)
|
|
run("ip", "link", "set", "lo", "up")
|
|
wait_for(lambda: "mv-server" in link_names())
|
|
run("ip", "addr", "add", "192.0.2.2/24", "dev", "mv-server")
|
|
run("ip", "-6", "addr", "add", "fd00:1::2/64", "dev", "mv-server", "nodad")
|
|
run("ip", "link", "set", "mv-server", "up")
|
|
run("ip", "link", "add", "remote", "type", "dummy")
|
|
run("ip", "addr", "add", "203.0.113.2/32", "dev", "remote")
|
|
run("ip", "addr", "add", "1.1.1.1/32", "dev", "remote")
|
|
run("ip", "-6", "addr", "add", "2001:db8:2::2/128", "dev", "remote", "nodad")
|
|
run("ip", "link", "set", "remote", "up")
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
data = self.client_address[0].encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
def log_message(self, *args):
|
|
pass
|
|
class IPv6Server(http.server.ThreadingHTTPServer):
|
|
address_family = socket.AF_INET6
|
|
for server in [http.server.ThreadingHTTPServer(("0.0.0.0", 8080), Handler), IPv6Server(("::", 8082), Handler)]:
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
class DoH(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
query = self.rfile.read(int(self.headers['Content-Length']))
|
|
proxy = self.client_address[0] != '192.0.2.1'
|
|
# Different answers prove which outbound the DNS exchange used.
|
|
answer = query[:2] + b'\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00' + query[12:] + b'\xc0\x0c\x00\x01\x00\x01\x00\x00\x00\x00\x00\x04' + bytes([203,0,113,10 if proxy else 20])
|
|
self.send_response(200)
|
|
self.send_header('Content-Type','application/dns-message')
|
|
self.send_header('Content-Length',str(len(answer)))
|
|
self.end_headers()
|
|
self.wfile.write(answer)
|
|
def log_message(self, *args): pass
|
|
doh = http.server.ThreadingHTTPServer(('1.1.1.1',443),DoH)
|
|
tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
tls.load_cert_chain(directory/'test-ca.pem',directory/'test-key.pem')
|
|
doh.socket = tls.wrap_socket(doh.socket,server_side=True)
|
|
threading.Thread(target=doh.serve_forever,daemon=True).start()
|
|
def udp(family, address):
|
|
s = socket.socket(family, socket.SOCK_DGRAM)
|
|
s.bind(address)
|
|
while True:
|
|
_, peer = s.recvfrom(1024)
|
|
s.sendto(peer[0].encode(), peer)
|
|
threading.Thread(target=udp, args=(socket.AF_INET,("0.0.0.0",8081)), daemon=True).start()
|
|
threading.Thread(target=udp, args=(socket.AF_INET6,("::",8083)), daemon=True).start()
|
|
cfg = {"log":{"level":"info"},"inbounds":[{"type":"vless","listen":"::","listen_port":2443,"users":[{"uuid":"b831381d-6324-4d53-ad4f-8cda48b30811"}]}],"outbounds":[{"type":"direct","tag":"direct"}],"route":{"final":"direct"}}
|
|
config = directory / "server.json"
|
|
config.write_text(json.dumps(cfg))
|
|
with (directory / "server.log").open("w") as log:
|
|
core = subprocess.Popen([str(CORE),"run","--disable-color","-c",str(config)],stdout=log,stderr=log)
|
|
try:
|
|
wait_for(lambda: "sing-box started" in (directory/"server.log").read_text())
|
|
(directory/"server-ready").touch()
|
|
while True:
|
|
time.sleep(1)
|
|
finally:
|
|
core.terminate()
|
|
core.wait(timeout=5)
|
|
|
|
def main():
|
|
if os.geteuid() != 0 or len(Path("/proc/self/uid_map").read_text().split()) != 3 or Path("/proc/self/uid_map").read_text().split()[2] != "1":
|
|
raise SystemExit("Run in an isolated unshare --user --map-root-user --net namespace, not as host root")
|
|
if link_names() != {"lo"}:
|
|
raise SystemExit("Refusing to change a nonempty network namespace")
|
|
with tempfile.TemporaryDirectory(prefix="minivless-test-") as tmp:
|
|
d = Path(tmp)
|
|
run('openssl','req','-x509','-newkey','rsa:2048','-nodes','-days','1','-subj','/CN=cloudflare-dns.com','-addext','subjectAltName=DNS:cloudflare-dns.com','-keyout',str(d/'test-key.pem'),'-out',str(d/'test-ca.pem'))
|
|
server = subprocess.Popen(["unshare","--net",sys.executable,__file__,"--server",tmp],start_new_session=True)
|
|
controllers = []
|
|
try:
|
|
run("ip","link","set","lo","up")
|
|
run("ip","link","add","mv-client","type","veth","peer","name","mv-server")
|
|
wait_for(lambda: os.readlink(f"/proc/{server.pid}/ns/net") != os.readlink("/proc/self/ns/net"))
|
|
run("ip","link","set","mv-server","netns",str(server.pid))
|
|
run("ip","addr","add","192.0.2.1/24","dev","mv-client")
|
|
run("ip","-6","addr","add","fd00:1::1/64","dev","mv-client","nodad")
|
|
run("ip","link","set","mv-client","up")
|
|
run("ip","route","add","default","via",SERVER_IP)
|
|
run("ip","-6","route","add","default","via","fd00:1::2")
|
|
wait_for(lambda:(d/"server-ready").exists())
|
|
clients = []
|
|
for name in ["selected-client","direct-client"]:
|
|
p = d/name
|
|
shutil.copyfile(DRIVER,p)
|
|
p.chmod(0o755)
|
|
clients.append(p)
|
|
baseline4 = run("ip","rule","show")
|
|
baseline6 = run("ip","-6","rule","show")
|
|
baseline_nft = run("nft","list","tables")
|
|
def start(mode):
|
|
prefs = d/f"config-{mode}"
|
|
prefs.mkdir(exist_ok=True,mode=0o700)
|
|
log = (d/f"controller-{mode}.log").open("w+")
|
|
p = subprocess.Popen([str(DRIVER),"--connect",str(CORE),str(prefs),str(clients[0]),mode],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=log,text=True,env={**os.environ,'SSL_CERT_FILE':str(d/'test-ca.pem')})
|
|
controllers.append(p)
|
|
if not select.select([p.stdout],[],[],22)[0] or p.stdout.readline().strip()!="READY":
|
|
log.seek(0)
|
|
raise AssertionError("Tunnel did not start:\n"+log.read())
|
|
return p,prefs
|
|
def clean():
|
|
return "minivless0" not in link_names() and run("ip","rule","show")==baseline4 and run("ip","-6","rule","show")==baseline6 and run("nft","list","tables")==baseline_nft
|
|
def check(mode):
|
|
for index,client in enumerate(clients):
|
|
proxy = mode=="full" or index==0
|
|
for flag,addr,expected_proxy,expected_direct in [
|
|
("--fetch","203.0.113.2:8080","203.0.113.2","192.0.2.1"),
|
|
("--udp","203.0.113.2:8081","203.0.113.2","192.0.2.1"),
|
|
("--fetch","[2001:db8:2::2]:8082","2001:db8:2::2","fd00:1::1"),
|
|
("--udp","[2001:db8:2::2]:8083","2001:db8:2::2","fd00:1::1"),
|
|
]:
|
|
peer=run(str(client),flag,addr)
|
|
expected=expected_proxy if proxy else expected_direct
|
|
assert peer==expected,(mode,client.name,flag,addr,peer,expected)
|
|
print(f"PASS {mode} {client.name} {flag} {addr}: {'VLESS' if proxy else 'DIRECT'}",flush=True)
|
|
answer=run(str(client),'--dns','192.0.2.2:53')
|
|
assert answer == ('203.0.113.10' if proxy else '203.0.113.20'), (mode, client.name, 'DNS', answer)
|
|
print(f"PASS {mode} {client.name} DNS over HTTPS: {'VLESS' if proxy else 'DIRECT'}",flush=True)
|
|
p,prefs=start("split")
|
|
check("split")
|
|
p.stdin.close();p.wait(timeout=10)
|
|
assert p.returncode==0
|
|
wait_for(clean)
|
|
assert not list(prefs.glob("runtime-*"))
|
|
print("PASS graceful shutdown restores IPv4/IPv6 routes and nftables",flush=True)
|
|
p,prefs=start("full")
|
|
check("full")
|
|
p.kill();p.wait(timeout=5)
|
|
wait_for(clean)
|
|
wait_for(lambda:not list(prefs.glob("runtime-*")))
|
|
print("PASS GUI SIGKILL: guardian stops core and cleans routes/config",flush=True)
|
|
finally:
|
|
for p in controllers:
|
|
if p.poll() is None:
|
|
p.stdin.close()
|
|
try:p.wait(timeout=8)
|
|
except subprocess.TimeoutExpired:p.kill();p.wait()
|
|
os.killpg(server.pid,signal.SIGTERM)
|
|
server.wait(timeout=5)
|
|
|
|
if __name__=="__main__":
|
|
if len(sys.argv)>1 and sys.argv[1]=="--server":serve(sys.argv[2])
|
|
else:main()
|