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