65 lines
3.2 KiB
Python
65 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Run an isolated SOCKS client with strict TLS; never print QA credentials."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--qa', required=True)
|
|
p.add_argument('--binary', required=True)
|
|
p.add_argument('--expected-exit', required=True)
|
|
p.add_argument('--port', type=int, default=19080)
|
|
p.add_argument('--subscription')
|
|
a = p.parse_args()
|
|
q = json.loads(pathlib.Path(a.qa).read_text())
|
|
c = {'log': {'level': 'error'}, 'dns': {'servers': [{'type': 'local', 'tag': 'local'}]},
|
|
'inbounds': [{'type': 'mixed', 'listen': '127.0.0.1', 'listen_port': a.port}],
|
|
'outbounds': [{'type': 'anytls', 'tag': 'qa', 'server': q['host'], 'server_port': q['port'],
|
|
'password': q['uuid'], 'tls': {'enabled': True, 'server_name': q['sni'], 'insecure': False}}],
|
|
'route': {'default_domain_resolver': 'local', 'final': 'qa'}}
|
|
if a.subscription:
|
|
c = json.loads(pathlib.Path(a.subscription).read_text())
|
|
nodes = [x for x in c['outbounds'] if x.get('type') == 'anytls']
|
|
assert len(nodes) == 1 and nodes[0]['tls'].get('insecure') is False
|
|
assert nodes[0]['server'] == q['host'] and nodes[0]['server_port'] == q['port']
|
|
assert nodes[0]['password'] == q['uuid']
|
|
c['inbounds'] = [{'type': 'mixed', 'listen': '127.0.0.1', 'listen_port': a.port}]
|
|
c['log'] = {'level': 'error'}
|
|
os.umask(0o077)
|
|
with tempfile.TemporaryDirectory(prefix='dpl-client-') as d:
|
|
config = pathlib.Path(d) / 'client.json'
|
|
config.write_text(json.dumps(c))
|
|
env = dict(os.environ, GOMEMLIMIT='64MiB', GOGC='75')
|
|
with (pathlib.Path(d) / 'client.log').open('wb') as log:
|
|
proc = subprocess.Popen([a.binary, 'run', '-c', str(config)], cwd=d, env=env, stdout=log, stderr=log)
|
|
try:
|
|
for _ in range(50):
|
|
assert proc.poll() is None, 'Client exited during startup'
|
|
try:
|
|
with socket.create_connection(('127.0.0.1', a.port), 0.1):
|
|
break
|
|
except OSError:
|
|
time.sleep(0.1)
|
|
command = ['curl', '-q', '--fail', '--silent', '--show-error', '--max-time', '30',
|
|
'--socks5-hostname', f'127.0.0.1:{a.port}']
|
|
result = subprocess.run(command + ['https://api.ipify.org'], capture_output=True, text=True, check=True)
|
|
ip = result.stdout.strip()
|
|
assert ip == a.expected_exit, 'Unexpected exit IP'
|
|
result = subprocess.run(command + ['https://www.cloudflare.com/cdn-cgi/trace'], capture_output=True, text=True, check=True)
|
|
trace = dict(line.split('=', 1) for line in result.stdout.splitlines() if '=' in line)
|
|
assert trace.get('ip') == a.expected_exit, 'Second exit check mismatch'
|
|
print(json.dumps({'anytls': True, 'relay': q['host'], 'port': q['port'], 'strict_tls': True,
|
|
'exit_ip': ip, 'country': trace.get('loc'), 'https_tls': trace.get('tls')}, ensure_ascii=False))
|
|
finally:
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
proc.wait()
|