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