Python code coverage for Lib/test/ssl_servers.py

#	count	content
1	n/a	import os
2	n/a	import sys
3	n/a	import ssl
4	n/a	import pprint
5	n/a	import socket
6	n/a	import urllib.parse
7	n/a	# Rename HTTPServer to _HTTPServer so as to avoid confusion with HTTPSServer.
8	n/a	from http.server import (HTTPServer as _HTTPServer,
9	n/a	SimpleHTTPRequestHandler, BaseHTTPRequestHandler)
10	n/a
11	n/a	from test import support
12	n/a	threading = support.import_module("threading")
13	n/a
14	n/a	here = os.path.dirname(__file__)
15	n/a
16	n/a	HOST = support.HOST
17	n/a	CERTFILE = os.path.join(here, 'keycert.pem')
18	n/a
19	n/a	# This one's based on HTTPServer, which is based on socketserver
20	n/a
21	n/a	class HTTPSServer(_HTTPServer):
22	n/a
23	n/a	def __init__(self, server_address, handler_class, context):
24	n/a	_HTTPServer.__init__(self, server_address, handler_class)
25	n/a	self.context = context
26	n/a
27	n/a	def __str__(self):
28	n/a	return ('<%s %s:%s>' %
29	n/a	(self.__class__.__name__,
30	n/a	self.server_name,
31	n/a	self.server_port))
32	n/a
33	n/a	def get_request(self):
34	n/a	# override this to wrap socket with SSL
35	n/a	try:
36	n/a	sock, addr = self.socket.accept()
37	n/a	sslconn = self.context.wrap_socket(sock, server_side=True)
38	n/a	except OSError as e:
39	n/a	# socket errors are silenced by the caller, print them here
40	n/a	if support.verbose:
41	n/a	sys.stderr.write("Got an error:\n%s\n" % e)
42	n/a	raise
43	n/a	return sslconn, addr
44	n/a
45	n/a	class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
46	n/a	# need to override translate_path to get a known root,
47	n/a	# instead of using os.curdir, since the test could be
48	n/a	# run from anywhere
49	n/a
50	n/a	server_version = "TestHTTPS/1.0"
51	n/a	root = here
52	n/a	# Avoid hanging when a request gets interrupted by the client
53	n/a	timeout = 5
54	n/a
55	n/a	def translate_path(self, path):
56	n/a	"""Translate a /-separated PATH to the local filename syntax.
57	n/a
58	n/a	Components that mean special things to the local file system
59	n/a	(e.g. drive or directory names) are ignored. (XXX They should
60	n/a	probably be diagnosed.)
61	n/a
62	n/a	"""
63	n/a	# abandon query parameters
64	n/a	path = urllib.parse.urlparse(path)[2]
65	n/a	path = os.path.normpath(urllib.parse.unquote(path))
66	n/a	words = path.split('/')
67	n/a	words = filter(None, words)
68	n/a	path = self.root
69	n/a	for word in words:
70	n/a	drive, word = os.path.splitdrive(word)
71	n/a	head, word = os.path.split(word)
72	n/a	path = os.path.join(path, word)
73	n/a	return path
74	n/a
75	n/a	def log_message(self, format, *args):
76	n/a	# we override this to suppress logging unless "verbose"
77	n/a	if support.verbose:
78	n/a	sys.stdout.write(" server (%s:%d %s):\n [%s] %s\n" %
79	n/a	(self.server.server_address,
80	n/a	self.server.server_port,
81	n/a	self.request.cipher(),
82	n/a	self.log_date_time_string(),
83	n/a	format%args))
84	n/a
85	n/a
86	n/a	class StatsRequestHandler(BaseHTTPRequestHandler):
87	n/a	"""Example HTTP request handler which returns SSL statistics on GET
88	n/a	requests.
89	n/a	"""
90	n/a
91	n/a	server_version = "StatsHTTPS/1.0"
92	n/a
93	n/a	def do_GET(self, send_body=True):
94	n/a	"""Serve a GET request."""
95	n/a	sock = self.rfile.raw._sock
96	n/a	context = sock.context
97	n/a	stats = {
98	n/a	'session_cache': context.session_stats(),
99	n/a	'cipher': sock.cipher(),
100	n/a	'compression': sock.compression(),
101	n/a	}
102	n/a	body = pprint.pformat(stats)
103	n/a	body = body.encode('utf-8')
104	n/a	self.send_response(200)
105	n/a	self.send_header("Content-type", "text/plain; charset=utf-8")
106	n/a	self.send_header("Content-Length", str(len(body)))
107	n/a	self.end_headers()
108	n/a	if send_body:
109	n/a	self.wfile.write(body)
110	n/a
111	n/a	def do_HEAD(self):
112	n/a	"""Serve a HEAD request."""
113	n/a	self.do_GET(send_body=False)
114	n/a
115	n/a	def log_request(self, format, *args):
116	n/a	if support.verbose:
117	n/a	BaseHTTPRequestHandler.log_request(self, format, *args)
118	n/a
119	n/a
120	n/a	class HTTPSServerThread(threading.Thread):
121	n/a
122	n/a	def __init__(self, context, host=HOST, handler_class=None):
123	n/a	self.flag = None
124	n/a	self.server = HTTPSServer((host, 0),
125	n/a	handler_class or RootedHTTPRequestHandler,
126	n/a	context)
127	n/a	self.port = self.server.server_port
128	n/a	threading.Thread.__init__(self)
129	n/a	self.daemon = True
130	n/a
131	n/a	def __str__(self):
132	n/a	return "<%s %s>" % (self.__class__.__name__, self.server)
133	n/a
134	n/a	def start(self, flag=None):
135	n/a	self.flag = flag
136	n/a	threading.Thread.start(self)
137	n/a
138	n/a	def run(self):
139	n/a	if self.flag:
140	n/a	self.flag.set()
141	n/a	try:
142	n/a	self.server.serve_forever(0.05)
143	n/a	finally:
144	n/a	self.server.server_close()
145	n/a
146	n/a	def stop(self):
147	n/a	self.server.shutdown()
148	n/a
149	n/a
150	n/a	def make_https_server(case, *, context=None, certfile=CERTFILE,
151	n/a	host=HOST, handler_class=None):
152	n/a	if context is None:
153	n/a	context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
154	n/a	# We assume the certfile contains both private key and certificate
155	n/a	context.load_cert_chain(certfile)
156	n/a	server = HTTPSServerThread(context, host, handler_class)
157	n/a	flag = threading.Event()
158	n/a	server.start(flag)
159	n/a	flag.wait()
160	n/a	def cleanup():
161	n/a	if support.verbose:
162	n/a	sys.stdout.write('stopping HTTPS server\n')
163	n/a	server.stop()
164	n/a	if support.verbose:
165	n/a	sys.stdout.write('joining HTTPS thread\n')
166	n/a	server.join()
167	n/a	case.addCleanup(cleanup)
168	n/a	return server
169	n/a
170	n/a
171	n/a	if __name__ == "__main__":
172	n/a	import argparse
173	n/a	parser = argparse.ArgumentParser(
174	n/a	description='Run a test HTTPS server. '
175	n/a	'By default, the current directory is served.')
176	n/a	parser.add_argument('-p', '--port', type=int, default=4433,
177	n/a	help='port to listen on (default: %(default)s)')
178	n/a	parser.add_argument('-q', '--quiet', dest='verbose', default=True,
179	n/a	action='store_false', help='be less verbose')
180	n/a	parser.add_argument('-s', '--stats', dest='use_stats_handler', default=False,
181	n/a	action='store_true', help='always return stats page')
182	n/a	parser.add_argument('--curve-name', dest='curve_name', type=str,
183	n/a	action='store',
184	n/a	help='curve name for EC-based Diffie-Hellman')
185	n/a	parser.add_argument('--ciphers', dest='ciphers', type=str,
186	n/a	help='allowed cipher list')
187	n/a	parser.add_argument('--dh', dest='dh_file', type=str, action='store',
188	n/a	help='PEM file containing DH parameters')
189	n/a	args = parser.parse_args()
190	n/a
191	n/a	support.verbose = args.verbose
192	n/a	if args.use_stats_handler:
193	n/a	handler_class = StatsRequestHandler
194	n/a	else:
195	n/a	handler_class = RootedHTTPRequestHandler
196	n/a	handler_class.root = os.getcwd()
197	n/a	context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
198	n/a	context.load_cert_chain(CERTFILE)
199	n/a	if args.curve_name:
200	n/a	context.set_ecdh_curve(args.curve_name)
201	n/a	if args.dh_file:
202	n/a	context.load_dh_params(args.dh_file)
203	n/a	if args.ciphers:
204	n/a	context.set_ciphers(args.ciphers)
205	n/a
206	n/a	server = HTTPSServer(("", args.port), handler_class, context)
207	n/a	if args.verbose:
208	n/a	print("Listening on https://localhost:{0.port}".format(args))
209	n/a	server.serve_forever(0.1)