188 lines
6.2 KiB
Python
188 lines
6.2 KiB
Python
import os
|
|
import re
|
|
import socket
|
|
from threading import Thread
|
|
|
|
import pytest
|
|
|
|
from pyhttpd.conf import HttpdConf
|
|
from pyhttpd.result import ExecResult
|
|
|
|
|
|
class UDSFaker:
|
|
|
|
def __init__(self, path):
|
|
self._uds_path = path
|
|
self._done = False
|
|
|
|
def start(self):
|
|
def process(self):
|
|
self._socket.listen(1)
|
|
self._process()
|
|
|
|
try:
|
|
os.unlink(self._uds_path)
|
|
except OSError:
|
|
if os.path.exists(self._uds_path):
|
|
raise
|
|
self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self._socket.bind(self._uds_path)
|
|
self._thread = Thread(target=process, daemon=True, args=[self])
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
self._done = True
|
|
self._socket.close()
|
|
|
|
def _process(self):
|
|
while self._done is False:
|
|
try:
|
|
c, client_address = self._socket.accept()
|
|
try:
|
|
data = c.recv(16)
|
|
c.sendall("""HTTP/1.1 200 Ok
|
|
Server: UdsFaker
|
|
Content-Type: application/json
|
|
Content-Length: 19
|
|
|
|
{ "host": "faked" }""".encode())
|
|
finally:
|
|
c.close()
|
|
|
|
except ConnectionAbortedError:
|
|
self._done = True
|
|
|
|
|
|
class TestProxyUds:
|
|
|
|
@pytest.fixture(autouse=True, scope='class')
|
|
def _class_scope(self, env):
|
|
# setup 3 vhosts on https: for reverse, forward and
|
|
# mixed proxying to a unix: domain socket
|
|
# We setup a UDSFaker running that returns a fixed response
|
|
UDS_PATH = f"{env.gen_dir}/proxy_02.sock"
|
|
TestProxyUds.UDS_PATH = UDS_PATH
|
|
faker = UDSFaker(path=UDS_PATH)
|
|
faker.start()
|
|
|
|
conf = HttpdConf(env)
|
|
conf.add("ProxyPreserveHost on")
|
|
conf.start_vhost(domains=[env.d_reverse], port=env.https_port)
|
|
conf.add([
|
|
f"ProxyPass / unix:{UDS_PATH}|http://127.0.0.1:{env.http_port}/"
|
|
])
|
|
conf.end_vhost()
|
|
|
|
conf.start_vhost(domains=[env.d_forward], port=env.https_port)
|
|
conf.add([
|
|
"ProxyRequests on"
|
|
])
|
|
conf.end_vhost()
|
|
|
|
conf.start_vhost(domains=[env.d_mixed], port=env.https_port)
|
|
conf.add([
|
|
f"ProxyPass / unix:{UDS_PATH}|http://127.0.0.1:{env.http_port}/",
|
|
"ProxyRequests on"
|
|
])
|
|
conf.end_vhost()
|
|
conf.install()
|
|
assert env.apache_restart() == 0
|
|
yield
|
|
faker.stop()
|
|
|
|
@pytest.mark.parametrize(["via", "seen"], [
|
|
["reverse", "faked"],
|
|
["mixed", "faked"],
|
|
])
|
|
def test_proxy_02_001(self, env, via, seen):
|
|
# make requests to a reverse proxy https: vhost to the http: vhost
|
|
# check that we see the document we expect there (host matching worked)
|
|
r = env.curl_get(f"https://{via}.{env.http_tld}:{env.https_port}/alive.json", 5)
|
|
assert r.response["status"] == 200
|
|
assert r.json['host'] == seen
|
|
|
|
@pytest.mark.parametrize(["via", "seen"], [
|
|
["forward", "generic"],
|
|
["mixed", "faked"],
|
|
])
|
|
def test_proxy_02_002(self, env, via, seen):
|
|
# make requests to a forward proxy https: vhost to the http: vhost
|
|
# check that we see the document we expect there (host matching worked)
|
|
# we need to explicitly provide a Host: header since mod_proxy cannot
|
|
# resolve the name via DNS.
|
|
if not env.curl_is_at_least('8.0.0'):
|
|
pytest.skip(f'need at least curl v8.0.0 for this')
|
|
domain = f"{via}.{env.http_tld}"
|
|
r = env.curl_get(f"http://127.0.0.1:{env.http_port}/alive.json", 5, options=[
|
|
'-H', f"Host: {domain}",
|
|
'--proxy', f"https://{domain}:{env.https_port}/",
|
|
'--resolve', f"{domain}:{env.https_port}:127.0.0.1",
|
|
'--proxy-cacert', f"{env.get_ca_pem_file(domain)}",
|
|
|
|
])
|
|
assert r.exit_code == 0, f"{r.stdout}{r.stderr}"
|
|
assert r.response["status"] == 200
|
|
assert r.json['host'] == seen
|
|
|
|
@pytest.mark.parametrize(["via", "exp_status"], [
|
|
["reverse", 400],
|
|
["forward", 500],
|
|
["mixed", 500],
|
|
])
|
|
def test_proxy_02_003(self, env, via, exp_status):
|
|
# make requests to a forward proxy https: vhost and GET
|
|
# a URL which carries the unix: domain socket.
|
|
# This needs to fail.
|
|
domain = f"{via}.{env.http_tld}"
|
|
r = env.run(args=[
|
|
'openssl', 's_client', '-connect', f"127.0.0.1:{env.https_port}",
|
|
'-servername', domain,
|
|
'-crlf', '-ign_eof',
|
|
'-CAfile', env.get_ca_pem_file(domain)
|
|
], intext=f"""GET unix:{TestProxyUds.UDS_PATH}|http://127.0.0.1:{env.http_port}/alive.json HTTP/1.1
|
|
Host: {domain}
|
|
|
|
""")
|
|
assert r.exit_code == 0, f"{r.stdout}{r.stderr}"
|
|
lines = r.stdout.split('\n')
|
|
rlines = None
|
|
for idx, l in enumerate(lines):
|
|
if l.startswith('HTTP/'):
|
|
rlines = lines[idx:]
|
|
assert rlines, f"No response found in: {r.stdout}"
|
|
r2 = self.parse_response(rlines)
|
|
assert r2.response
|
|
assert r2.response['status'] == exp_status
|
|
|
|
def parse_response(self, lines) -> ExecResult:
|
|
exp_body = False
|
|
exp_stat = True
|
|
r = ExecResult(args=[], exit_code=0, stdout=b'', stderr=b'')
|
|
header = {}
|
|
body = []
|
|
for line in lines:
|
|
if exp_stat:
|
|
m = re.match(r'^(\S+) (\d+) (.*)$', line)
|
|
assert m, f"first line no HTTP status line: {line}"
|
|
r.add_response({
|
|
"protocol": m.group(1),
|
|
"status": int(m.group(2)),
|
|
"description": m.group(3),
|
|
"body": r.outraw
|
|
})
|
|
header = {}
|
|
exp_stat = False
|
|
exp_body = False
|
|
elif re.match(r'^\r?$', line):
|
|
exp_body = True
|
|
elif exp_body:
|
|
body.append(line)
|
|
else:
|
|
m = re.match(r'^([^:]+):\s*(.*)$', line)
|
|
assert m, f"not a header line: {line}"
|
|
header[m.group(1).lower()] = m.group(2)
|
|
if r.response:
|
|
r.response["header"] = header
|
|
r.response["body"] = body
|
|
return r
|