113 lines
5.1 KiB
Python
113 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise real Nginx mapping and Origin/Referer isolation using localhost only."""
|
|
import importlib.util
|
|
import pathlib
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.request
|
|
|
|
spec = importlib.util.spec_from_file_location('render', pathlib.Path(__file__).with_name('render.py'))
|
|
render = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(render)
|
|
nginx = shutil.which('nginx')
|
|
if not nginx:
|
|
raise SystemExit('Nginx is required for this isolated integration check')
|
|
|
|
with tempfile.TemporaryDirectory(prefix='gateway-proxy-maps-') as folder:
|
|
root = pathlib.Path(folder)
|
|
with socket.socket() as sock:
|
|
sock.bind(('127.0.0.1', 0))
|
|
port = sock.getsockname()[1]
|
|
maps = render.render_white('https', '/unused')['00-gateway-admin-proxy-maps.conf']
|
|
bodies = [
|
|
('weiwei', 'weiwei.tomtojapan.online', 'hw.lshan.eu.org', 'weiwei.tomtojapan.online', 'weiwei', '海外转发'),
|
|
('dpl', 'dpl.tomtojapan.online', 'dp-link.net', 'dpl.tomtojapan.online', 'dpl-site', 'DP-LINK.NET'),
|
|
('fuc', 'fuc.tomtojapan.online', 'xn--9kqy16h.com', 'fuc.tomtojapan.online', 'fuc-site', '福云.com'),
|
|
('new', 'getway-dpl.tomshell.online', 'gemcook.net', 'tomshell.online', 'new', 'GEMCOOK.NET'),
|
|
('old', 'nasa.tomshell.com.de', 'diaodu.us', 'tomshell.com.de', 'old', 'DIAODU.US'),
|
|
]
|
|
locations = ''
|
|
for name, host, source, alias, key, brand in bodies:
|
|
upstream = ('getway-dpl.gemcook.net' if key == 'new' else
|
|
'nasa.diaodu.us' if key == 'old' else source)
|
|
body = f'<title>{brand}</title><a href="https://{upstream}/api?q=1">link</a><img src="//{upstream}/logo">'
|
|
locations += f''' location /body/{name} {{
|
|
default_type text/html;
|
|
{render.body_rewrites(source, alias, key)}
|
|
return 200 '{body}';
|
|
}}
|
|
'''
|
|
config = root / 'nginx.conf'
|
|
config.write_text(f'''daemon off;
|
|
pid {root}/nginx.pid;
|
|
error_log {root}/error.log;
|
|
events {{ worker_connections 64; }}
|
|
http {{
|
|
access_log off;
|
|
{maps}
|
|
server {{
|
|
listen 127.0.0.1:{port};
|
|
location / {{
|
|
return 200 "$gateway_admin_origin_host|$gateway_admin_origin|$gateway_admin_referer";
|
|
}}
|
|
{locations}
|
|
}}
|
|
}}
|
|
''')
|
|
subprocess.run([nginx, '-t', '-c', str(config), '-p', folder], check=True,
|
|
capture_output=True, text=True)
|
|
process = subprocess.Popen([nginx, '-c', str(config), '-p', folder],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
for _ in range(40):
|
|
try:
|
|
with socket.create_connection(('127.0.0.1', port), timeout=.1):
|
|
break
|
|
except OSError:
|
|
if process.poll() is not None:
|
|
raise RuntimeError(process.stderr.read())
|
|
time.sleep(.05)
|
|
cases = [
|
|
('tomshell.online', 'gemcook.net'),
|
|
('getway-fuc.tomshell.online', 'getway-fuc.gemcook.net'),
|
|
('getway-dpl.tomshell.online', 'getway-dpl.gemcook.net'),
|
|
('nasa.tomshell.com.de', 'nasa.diaodu.us'),
|
|
('tomshell.com.de', 'diaodu.us'),
|
|
('dpl.tomtojapan.online', 'dp-link.net'),
|
|
('fuc.tomtojapan.online', 'xn--9kqy16h.com'),
|
|
('not-registered.example', ''),
|
|
('two.levels.tomshell.online', ''),
|
|
]
|
|
assertions = 0
|
|
for host, upstream in cases:
|
|
for origin in ['https://' + host, 'https://other.tomshell.online', 'https://outside.example']:
|
|
referer = origin + '/example?keep=1'
|
|
req = urllib.request.Request('http://127.0.0.1:' + str(port) + '/',
|
|
headers={'Host': host, 'Origin': origin, 'Referer': referer})
|
|
with urllib.request.urlopen(req, timeout=3) as response:
|
|
actual = response.read().decode().split('|')
|
|
assert actual[0] == upstream, (host, actual[0], upstream)
|
|
own = origin == 'https://' + host
|
|
assert actual[1] == ('https://' + upstream if own else origin), (host, origin, actual[1])
|
|
assert actual[2] == ('https://' + upstream + '/example?keep=1' if own else referer)
|
|
assertions += 3
|
|
for name, host, source, alias, key, brand in bodies:
|
|
req = urllib.request.Request(f'http://127.0.0.1:{port}/body/{name}', headers={'Host': host})
|
|
with urllib.request.urlopen(req, timeout=3) as response:
|
|
actual = response.read().decode()
|
|
assert f'<title>{brand}</title>' in actual, (name, actual)
|
|
assert f'https://{host}/api?q=1' in actual, (name, actual)
|
|
assert f'//{host}/logo' in actual, (name, actual)
|
|
assertions += 3
|
|
print(f'PASS: {assertions} real-Nginx map/header checks; localhost fixture only, no origin requests')
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|