diff options
Diffstat (limited to 'test/modules/proxy/test_02_unix.py')
| -rw-r--r-- | test/modules/proxy/test_02_unix.py | 187 |
1 files changed, 187 insertions, 0 deletions
diff --git a/test/modules/proxy/test_02_unix.py b/test/modules/proxy/test_02_unix.py new file mode 100644 index 0000000..7f3d4d5 --- /dev/null +++ b/test/modules/proxy/test_02_unix.py @@ -0,0 +1,187 @@ +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 |