1import inspect 2import logging 3import os 4import re 5import subprocess 6from typing import Dict, Any 7 8from pyhttpd.certs import CertificateSpec 9from pyhttpd.conf import HttpdConf 10from pyhttpd.env import HttpdTestEnv, HttpdTestSetup 11 12log = logging.getLogger(__name__) 13 14 15class H2TestSetup(HttpdTestSetup): 16 17 def __init__(self, env: 'HttpdTestEnv'): 18 super().__init__(env=env) 19 self.add_source_dir(os.path.dirname(inspect.getfile(H2TestSetup))) 20 self.add_modules(["http2", "proxy_http2", "cgid", "autoindex"]) 21 22 def make(self): 23 super().make() 24 self._add_h2test() 25 self._setup_data_1k_1m() 26 27 def _add_h2test(self): 28 local_dir = os.path.dirname(inspect.getfile(H2TestSetup)) 29 p = subprocess.run([self.env.apxs, '-c', 'mod_h2test.c'], 30 capture_output=True, 31 cwd=os.path.join(local_dir, 'mod_h2test')) 32 rv = p.returncode 33 if rv != 0: 34 log.error(f"compiling md_h2test failed: {p.stderr}") 35 raise Exception(f"compiling md_h2test failed: {p.stderr}") 36 37 modules_conf = os.path.join(self.env.server_dir, 'conf/modules.conf') 38 with open(modules_conf, 'a') as fd: 39 # load our test module which is not installed 40 fd.write(f"LoadModule h2test_module \"{local_dir}/mod_h2test/.libs/mod_h2test.so\"\n") 41 42 def _setup_data_1k_1m(self): 43 s90 = "01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678\n" 44 with open(os.path.join(self.env.gen_dir, "data-1k"), 'w') as f: 45 for i in range(10): 46 f.write(f"{i:09d}-{s90}") 47 with open(os.path.join(self.env.gen_dir, "data-10k"), 'w') as f: 48 for i in range(100): 49 f.write(f"{i:09d}-{s90}") 50 with open(os.path.join(self.env.gen_dir, "data-100k"), 'w') as f: 51 for i in range(1000): 52 f.write(f"{i:09d}-{s90}") 53 with open(os.path.join(self.env.gen_dir, "data-1m"), 'w') as f: 54 for i in range(10000): 55 f.write(f"{i:09d}-{s90}") 56 57 58class H2TestEnv(HttpdTestEnv): 59 60 def __init__(self, pytestconfig=None): 61 super().__init__(pytestconfig=pytestconfig) 62 self.add_httpd_conf([ 63 "H2MinWorkers 1", 64 "H2MaxWorkers 64", 65 "Protocols h2 http/1.1 h2c", 66 ]) 67 self.add_httpd_log_modules(["http2", "proxy_http2", "h2test"]) 68 self.add_cert_specs([ 69 CertificateSpec(domains=[ 70 f"push.{self._http_tld}", 71 f"hints.{self._http_tld}", 72 f"ssl.{self._http_tld}", 73 f"pad0.{self._http_tld}", 74 f"pad1.{self._http_tld}", 75 f"pad2.{self._http_tld}", 76 f"pad3.{self._http_tld}", 77 f"pad8.{self._http_tld}", 78 ]), 79 CertificateSpec(domains=[f"noh2.{self.http_tld}"], key_type='rsa2048'), 80 ]) 81 82 self.httpd_error_log.set_ignored_lognos([ 83 'AH02032', 84 'AH01276', 85 'AH01630', 86 'AH00135', 87 'AH02261', # Re-negotiation handshake failed (our test_101) 88 'AH03490', # scoreboard full, happens on limit tests 89 ]) 90 self.httpd_error_log.add_ignored_patterns([ 91 re.compile(r'.*malformed header from script \'hecho.py\': Bad header: x.*'), 92 re.compile(r'.*:tls_post_process_client_hello:.*'), 93 re.compile(r'.*:tls_process_client_certificate:.*'), 94 ]) 95 96 def setup_httpd(self, setup: HttpdTestSetup = None): 97 super().setup_httpd(setup=H2TestSetup(env=self)) 98 99 100class H2Conf(HttpdConf): 101 102 def __init__(self, env: HttpdTestEnv, extras: Dict[str, Any] = None): 103 super().__init__(env=env, extras=HttpdConf.merge_extras(extras, { 104 f"cgi.{env.http_tld}": [ 105 "SSLOptions +StdEnvVars", 106 "AddHandler cgi-script .py", 107 ] 108 })) 109 110 def start_vhost(self, domains, port=None, doc_root="htdocs", with_ssl=None): 111 super().start_vhost(domains=domains, port=port, doc_root=doc_root, with_ssl=with_ssl) 112 if f"noh2.{self.env.http_tld}" in domains: 113 protos = ["http/1.1"] 114 elif port == self.env.https_port or with_ssl is True: 115 protos = ["h2", "http/1.1"] 116 else: 117 protos = ["h2c", "http/1.1"] 118 if f"test2.{self.env.http_tld}" in domains: 119 protos = reversed(protos) 120 self.add(f"Protocols {' '.join(protos)}") 121 return self 122 123 def add_vhost_noh2(self): 124 domains = [f"noh2.{self.env.http_tld}", f"noh2-alias.{self.env.http_tld}"] 125 self.start_vhost(domains=domains, port=self.env.https_port, doc_root="htdocs/noh2") 126 self.add(["Protocols http/1.1", "SSLOptions +StdEnvVars"]) 127 self.end_vhost() 128 self.start_vhost(domains=domains, port=self.env.http_port, doc_root="htdocs/noh2") 129 self.add(["Protocols http/1.1", "SSLOptions +StdEnvVars"]) 130 self.end_vhost() 131 return self 132 133 def add_vhost_test1(self, proxy_self=False, h2proxy_self=False): 134 return super().add_vhost_test1(proxy_self=proxy_self, h2proxy_self=h2proxy_self) 135 136 def add_vhost_test2(self): 137 return super().add_vhost_test2() 138