1from typing import Dict, Any 2 3from pyhttpd.env import HttpdTestEnv 4 5 6class HttpdConf(object): 7 8 def __init__(self, env: HttpdTestEnv, extras: Dict[str, Any] = None): 9 """ Create a new httpd configuration. 10 :param env: then environment this operates in 11 :param extras: extra configuration directive with ServerName as key and 12 'base' as special key for global configuration additions. 13 """ 14 self.env = env 15 self._indents = 0 16 self._lines = [] 17 self._extras = extras.copy() if extras else {} 18 if 'base' in self._extras: 19 self.add(self._extras['base']) 20 21 def __repr__(self): 22 s = '\n'.join(self._lines) 23 return f"HttpdConf[{s}]" 24 25 def install(self): 26 self.env.install_test_conf(self._lines) 27 28 def add(self, line: Any): 29 if isinstance(line, str): 30 if self._indents > 0: 31 line = f"{' ' * self._indents}{line}" 32 self._lines.append(line) 33 else: 34 if self._indents > 0: 35 line = [f"{' ' * self._indents}{l}" for l in line] 36 self._lines.extend(line) 37 return self 38 39 def add_certificate(self, cert_file, key_file): 40 if self.env.ssl_module == "ssl": 41 self.add([ 42 f"SSLCertificateFile {cert_file}", 43 f"SSLCertificateKeyFile {key_file if key_file else cert_file}", 44 ]) 45 elif self.env.ssl_module == "tls": 46 self.add(f""" 47 TLSCertificate {cert_file} {key_file} 48 """) 49 50 def add_vhost(self, domains, port=None, doc_root="htdocs", with_ssl=None): 51 self.start_vhost(domains=domains, port=port, doc_root=doc_root, with_ssl=with_ssl) 52 self.end_vhost() 53 return self 54 55 def start_vhost(self, domains, port=None, doc_root="htdocs", with_ssl=None): 56 if not isinstance(domains, list): 57 domains = [domains] 58 if port is None: 59 port = self.env.https_port 60 if with_ssl is None: 61 with_ssl = (self.env.https_port == port) 62 self.add("") 63 self.add(f"<VirtualHost *:{port}>") 64 self._indents += 1 65 self.add(f"ServerName {domains[0]}") 66 for alias in domains[1:]: 67 self.add(f"ServerAlias {alias}") 68 self.add(f"DocumentRoot {doc_root}") 69 if with_ssl: 70 if self.env.ssl_module == "ssl": 71 self.add("SSLEngine on") 72 for cred in self.env.get_credentials_for_name(domains[0]): 73 self.add_certificate(cred.cert_file, cred.pkey_file) 74 if domains[0] in self._extras: 75 self.add(self._extras[domains[0]]) 76 return self 77 78 def end_vhost(self): 79 self._indents -= 1 80 self.add("</VirtualHost>") 81 self.add("") 82 return self 83 84 def add_proxies(self, host, proxy_self=False, h2proxy_self=False): 85 if proxy_self or h2proxy_self: 86 self.add("ProxyPreserveHost on") 87 if proxy_self: 88 self.add([ 89 f"ProxyPass /proxy/ http://127.0.0.1:{self.env.http_port}/", 90 f"ProxyPassReverse /proxy/ http://{host}.{self.env.http_tld}:{self.env.http_port}/", 91 ]) 92 if h2proxy_self: 93 self.add([ 94 f"ProxyPass /h2proxy/ h2://127.0.0.1:{self.env.https_port}/", 95 f"ProxyPassReverse /h2proxy/ https://{host}.{self.env.http_tld}:self.env.https_port/", 96 ]) 97 return self 98 99 def add_vhost_test1(self, proxy_self=False, h2proxy_self=False): 100 domain = f"test1.{self.env.http_tld}" 101 self.start_vhost(domains=[domain, f"www1.{self.env.http_tld}"], 102 port=self.env.http_port, doc_root="htdocs/test1") 103 self.end_vhost() 104 self.start_vhost(domains=[domain, f"www1.{self.env.http_tld}"], 105 port=self.env.https_port, doc_root="htdocs/test1") 106 self.add([ 107 "<Location /006>", 108 " Options +Indexes", 109 "</Location>", 110 ]) 111 self.add_proxies("test1", proxy_self, h2proxy_self) 112 self.end_vhost() 113 return self 114 115 def add_vhost_test2(self): 116 domain = f"test2.{self.env.http_tld}" 117 self.start_vhost(domains=[domain, f"www2.{self.env.http_tld}"], 118 port=self.env.http_port, doc_root="htdocs/test2") 119 self.end_vhost() 120 self.start_vhost(domains=[domain, f"www2.{self.env.http_tld}"], 121 port=self.env.https_port, doc_root="htdocs/test2") 122 self.add([ 123 "<Location /006>", 124 " Options +Indexes", 125 "</Location>", 126 ]) 127 self.end_vhost() 128 return self 129 130 def add_vhost_cgi(self, proxy_self=False, h2proxy_self=False): 131 domain = f"cgi.{self.env.http_tld}" 132 if proxy_self: 133 self.add(["ProxyStatus on", "ProxyTimeout 5", 134 "SSLProxyEngine on", "SSLProxyVerify none"]) 135 if h2proxy_self: 136 self.add(["SSLProxyEngine on", "SSLProxyCheckPeerName off"]) 137 self.start_vhost(domains=[domain, f"cgi-alias.{self.env.http_tld}"], 138 port=self.env.https_port, doc_root="htdocs/cgi") 139 self.add_proxies("cgi", proxy_self=proxy_self, h2proxy_self=h2proxy_self) 140 self.add("<Location \"/h2test/echo\">") 141 self.add(" SetHandler h2test-echo") 142 self.add("</Location>") 143 self.add("<Location \"/h2test/delay\">") 144 self.add(" SetHandler h2test-delay") 145 self.add("</Location>") 146 if domain in self._extras: 147 self.add(self._extras[domain]) 148 self.end_vhost() 149 self.start_vhost(domains=[domain, f"cgi-alias.{self.env.http_tld}"], 150 port=self.env.http_port, doc_root="htdocs/cgi") 151 self.add("AddHandler cgi-script .py") 152 self.add_proxies("cgi", proxy_self=proxy_self, h2proxy_self=h2proxy_self) 153 self.end_vhost() 154 self.add("LogLevel proxy:info") 155 self.add("LogLevel proxy_http:info") 156 return self 157 158 @staticmethod 159 def merge_extras(e1: Dict[str, Any], e2: Dict[str, Any]) -> Dict[str, Any]: 160 def _concat(v1, v2): 161 if isinstance(v1, str): 162 v1 = [v1] 163 if isinstance(v2, str): 164 v2 = [v2] 165 v1.extend(v2) 166 return v1 167 168 if e1 is None: 169 return e2.copy() if e2 else None 170 if e2 is None: 171 return e1.copy() 172 e3 = e1.copy() 173 for name, val in e2.items(): 174 if name in e3: 175 e3[name] = _concat(e3[name], val) 176 else: 177 e3[name] = val 178 return e3 179