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