httpd.py (6808B)
1 #!/usr/bin/env python 2 3 # This Source Code Form is subject to the terms of the Mozilla Public 4 # License, v. 2.0. If a copy of the MPL was not distributed with this 5 # file, You can obtain one at http://mozilla.org/MPL/2.0/. 6 7 """Specialisation of wptserver.server.WebTestHttpd for testing 8 Marionette. 9 10 """ 11 12 import argparse 13 import os 14 import select 15 import sys 16 import time 17 from urllib.parse import parse_qsl, urlparse 18 19 from wptserve import handlers, request, server 20 from wptserve import routes as default_routes 21 22 root = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) 23 default_doc_root = os.path.join(root, "www") 24 default_ssl_cert = os.path.join(root, "certificates", "test.cert") 25 default_ssl_key = os.path.join(root, "certificates", "test.key") 26 27 28 @handlers.handler 29 def http_auth_handler(req, response): 30 # Allow the test to specify the username and password 31 params = dict(parse_qsl(req.url_parts.query)) 32 username = params.get("username", "guest") 33 password = params.get("password", "guest") 34 35 auth = request.Authentication(req.headers) 36 content = """<!doctype html> 37 <title>HTTP Authentication</title> 38 <p id="status">{}</p>""" 39 40 if auth.username == username and auth.password == password: 41 response.status = 200 42 response.content = content.format("success") 43 44 else: 45 response.status = 401 46 response.headers.set("WWW-Authenticate", 'Basic realm="secret"') 47 response.content = content.format("restricted") 48 49 50 @handlers.handler 51 def upload_handler(request, response): 52 return 200, [], [request.headers.get("Content-Type")] or [] 53 54 55 @handlers.handler 56 def slow_loading_handler(request, response): 57 # Allow the test specify the delay for delivering the content 58 params = dict(parse_qsl(request.url_parts.query)) 59 delay = int(params.get("delay", 5)) 60 time.sleep(delay) 61 62 # Do not allow the page to be cached to circumvent the bfcache of the browser 63 response.headers.set("Cache-Control", "no-cache, no-store") 64 response.content = f"""<!doctype html> 65 <meta charset="UTF-8"> 66 <title>Slow page loading</title> 67 68 <p>Delay: <span id="delay">{delay}</span></p> 69 """ 70 71 72 @handlers.handler 73 def slow_coop_handler(request, response): 74 # Allow the test specify the delay for delivering the content 75 params = dict(parse_qsl(request.url_parts.query)) 76 delay = int(params.get("delay", 5)) 77 time.sleep(delay) 78 79 # Isolate the browsing context exclusively to same-origin documents 80 response.headers.set("Cross-Origin-Opener-Policy", "same-origin") 81 response.headers.set("Cache-Control", "no-cache, no-store") 82 response.content = f"""<!doctype html> 83 <meta charset="UTF-8"> 84 <title>Slow cross-origin page loading</title> 85 86 <p>Delay: <span id="delay">{delay}</span></p> 87 """ 88 89 90 @handlers.handler 91 def update_xml_handler(request, response): 92 response.headers.set("Content-Type", "text/xml") 93 mar_digest = ( 94 "75cd68e6c98c84c435cd27e353f5b4f6a3f2c50f6802aa9bf62b47e47138757306769fd9befa08793635ee649" 95 "2319253480860b4aa8ed9ee1caaa4c83ebc90b9" 96 ) 97 response.content = f""" 98 <updates> 99 <update type="minor" displayVersion="9999.0" appVersion="9999.0" platformVersion="9999.0" 100 buildID="20220627075547"> 101 <patch type="complete" URL="{request.url_parts.scheme}://{request.url_parts.netloc}/update/complete.mar" size="86612" 102 hashFunction="sha512" hashValue="{mar_digest}"/> 103 </update> 104 </updates> 105 """ 106 107 108 class NotAliveError(Exception): 109 """Occurs when attempting to run a function that requires the HTTPD 110 to have been started, and it has not. 111 112 """ 113 114 pass 115 116 117 class FixtureServer: 118 def __init__( 119 self, 120 doc_root, 121 url="http://127.0.0.1:0", 122 use_ssl=False, 123 ssl_cert=None, 124 ssl_key=None, 125 ): 126 if not os.path.isdir(doc_root): 127 raise ValueError("Server root is not a directory: %s" % doc_root) 128 129 url = urlparse(url) 130 if url.scheme is None: 131 raise ValueError("Server scheme not provided") 132 133 scheme, host, port = url.scheme, url.hostname, url.port 134 if host is None: 135 host = "127.0.0.1" 136 if port is None: 137 port = 0 138 139 routes = [ 140 ("POST", "/file_upload", upload_handler), 141 ("GET", "/http_auth", http_auth_handler), 142 ("GET", "/slow", slow_loading_handler), 143 ("GET", "/slow-coop", slow_coop_handler), 144 ("GET", "/update.xml", update_xml_handler), 145 ] 146 routes.extend(default_routes.routes) 147 148 self._httpd = server.WebTestHttpd( 149 host=host, 150 port=port, 151 bind_address=True, 152 doc_root=doc_root, 153 routes=routes, 154 use_ssl=True if scheme == "https" else False, 155 certificate=ssl_cert, 156 key_file=ssl_key, 157 ) 158 159 def start(self): 160 if self.is_alive: 161 return 162 self._httpd.start() 163 164 def wait(self): 165 if not self.is_alive: 166 return 167 try: 168 select.select([], [], []) 169 except KeyboardInterrupt: 170 self.stop() 171 172 def stop(self): 173 if not self.is_alive: 174 return 175 self._httpd.stop() 176 177 def get_url(self, path): 178 if not self.is_alive: 179 raise NotAliveError() 180 return self._httpd.get_url(path) 181 182 @property 183 def doc_root(self): 184 return self._httpd.router.doc_root 185 186 @property 187 def router(self): 188 return self._httpd.router 189 190 @property 191 def routes(self): 192 return self._httpd.router.routes 193 194 @property 195 def is_alive(self): 196 return self._httpd.started 197 198 199 if __name__ == "__main__": 200 parser = argparse.ArgumentParser( 201 description="Specialised HTTP server for testing Marionette." 202 ) 203 parser.add_argument( 204 "url", 205 help=""" 206 service address including scheme, hostname, port, and prefix for document root, 207 e.g. \"https://0.0.0.0:0/base/\"""", 208 ) 209 parser.add_argument( 210 "-r", 211 dest="doc_root", 212 default=default_doc_root, 213 help="path to document root (default %(default)s)", 214 ) 215 parser.add_argument( 216 "-c", 217 dest="ssl_cert", 218 default=default_ssl_cert, 219 help="path to SSL certificate (default %(default)s)", 220 ) 221 parser.add_argument( 222 "-k", 223 dest="ssl_key", 224 default=default_ssl_key, 225 help="path to SSL certificate key (default %(default)s)", 226 ) 227 args = parser.parse_args() 228 229 httpd = FixtureServer( 230 args.doc_root, args.url, ssl_cert=args.ssl_cert, ssl_key=args.ssl_key 231 ) 232 httpd.start() 233 print( 234 "{}: started fixture server on {}".format(sys.argv[0], httpd.get_url("/")), 235 file=sys.stderr, 236 ) 237 httpd.wait()