D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
2
/
task
/
2
/
root
/
opt
/
imh-python
/
lib
/
python3.9
/
site-packages
/
cherrypy
/
test
/
Filename :
test_conn.py
back
Copy
"""Tests for TCP connection handling, including proper and timely close.""" import errno import socket import sys import time import urllib.parse from http.client import BadStatusLine, HTTPConnection, NotConnected from cheroot.test import webtest import cherrypy from cherrypy._cpcompat import HTTPSConnection, ntob, tonative from cherrypy.test import helper timeout = 1 pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN' def setup_server(): def raise500(): raise cherrypy.HTTPError(500) class Root: @cherrypy.expose def index(self): return pov page1 = index page2 = index page3 = index @cherrypy.expose def hello(self): return 'Hello, world!' @cherrypy.expose def timeout(self, t): return str(cherrypy.server.httpserver.timeout) @cherrypy.expose @cherrypy.config(**{'response.stream': True}) def stream(self, set_cl=False): if set_cl: cherrypy.response.headers['Content-Length'] = 10 def content(): for x in range(10): yield str(x) return content() @cherrypy.expose def error(self, code=500): raise cherrypy.HTTPError(code) @cherrypy.expose def upload(self): if not cherrypy.request.method == 'POST': raise AssertionError("'POST' != request.method %r" % cherrypy.request.method) return "thanks for '%s'" % cherrypy.request.body.read() @cherrypy.expose def custom(self, response_code): cherrypy.response.status = response_code return 'Code = %s' % response_code @cherrypy.expose @cherrypy.config(**{'hooks.on_start_resource': raise500}) def err_before_read(self): return 'ok' @cherrypy.expose def one_megabyte_of_a(self): return ['a' * 1024] * 1024 @cherrypy.expose # Turn off the encoding tool so it doens't collapse # our response body and reclaculate the Content-Length. @cherrypy.config(**{'tools.encode.on': False}) def custom_cl(self, body, cl): cherrypy.response.headers['Content-Length'] = cl if not isinstance(body, list): body = [body] newbody = [] for chunk in body: if isinstance(chunk, str): chunk = chunk.encode('ISO-8859-1') newbody.append(chunk) return newbody cherrypy.tree.mount(Root()) cherrypy.config.update({ 'server.max_request_body_size': 1001, 'server.socket_timeout': timeout, }) class ConnectionCloseTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_HTTP11(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' self.persistent = True # Make the first request and assert there's no "Connection: close". self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) self.assertNoHeader('Connection') # Make another request on the same connection. self.getPage('/page1') self.assertStatus('200 OK') self.assertBody(pov) self.assertNoHeader('Connection') # Test client-side close. self.getPage('/page2', headers=[('Connection', 'close')]) self.assertStatus('200 OK') self.assertBody(pov) self.assertHeader('Connection', 'close') # Make another request on the same connection, which should error. self.assertRaises(NotConnected, self.getPage, '/') def test_Streaming_no_len(self): try: self._streaming(set_cl=False) finally: try: self.HTTP_CONN.close() except (TypeError, AttributeError): pass def test_Streaming_with_len(self): try: self._streaming(set_cl=True) finally: try: self.HTTP_CONN.close() except (TypeError, AttributeError): pass def _streaming(self, set_cl): if cherrypy.server.protocol_version == 'HTTP/1.1': self.PROTOCOL = 'HTTP/1.1' self.persistent = True # Make the first request and assert there's no "Connection: close". self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) self.assertNoHeader('Connection') # Make another, streamed request on the same connection. if set_cl: # When a Content-Length is provided, the content should stream # without closing the connection. self.getPage('/stream?set_cl=Yes') self.assertHeader('Content-Length') self.assertNoHeader('Connection', 'close') self.assertNoHeader('Transfer-Encoding') self.assertStatus('200 OK') self.assertBody('0123456789') else: # When no Content-Length response header is provided, # streamed output will either close the connection, or use # chunked encoding, to determine transfer-length. self.getPage('/stream') self.assertNoHeader('Content-Length') self.assertStatus('200 OK') self.assertBody('0123456789') chunked_response = False for k, v in self.headers: if k.lower() == 'transfer-encoding': if str(v) == 'chunked': chunked_response = True if chunked_response: self.assertNoHeader('Connection', 'close') else: self.assertHeader('Connection', 'close') # Make another request on the same connection, which should # error. self.assertRaises(NotConnected, self.getPage, '/') # Try HEAD. See # https://github.com/cherrypy/cherrypy/issues/864. self.getPage('/stream', method='HEAD') self.assertStatus('200 OK') self.assertBody('') self.assertNoHeader('Transfer-Encoding') else: self.PROTOCOL = 'HTTP/1.0' self.persistent = True # Make the first request and assert Keep-Alive. self.getPage('/', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody(pov) self.assertHeader('Connection', 'Keep-Alive') # Make another, streamed request on the same connection. if set_cl: # When a Content-Length is provided, the content should # stream without closing the connection. self.getPage('/stream?set_cl=Yes', headers=[('Connection', 'Keep-Alive')]) self.assertHeader('Content-Length') self.assertHeader('Connection', 'Keep-Alive') self.assertNoHeader('Transfer-Encoding') self.assertStatus('200 OK') self.assertBody('0123456789') else: # When a Content-Length is not provided, # the server should close the connection. self.getPage('/stream', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody('0123456789') self.assertNoHeader('Content-Length') self.assertNoHeader('Connection', 'Keep-Alive') self.assertNoHeader('Transfer-Encoding') # Make another request on the same connection, which should # error. self.assertRaises(NotConnected, self.getPage, '/') def test_HTTP10_KeepAlive(self): self.PROTOCOL = 'HTTP/1.0' if self.scheme == 'https': self.HTTP_CONN = HTTPSConnection else: self.HTTP_CONN = HTTPConnection # Test a normal HTTP/1.0 request. self.getPage('/page2') self.assertStatus('200 OK') self.assertBody(pov) # Apache, for example, may emit a Connection header even for HTTP/1.0 # self.assertNoHeader("Connection") # Test a keep-alive HTTP/1.0 request. self.persistent = True self.getPage('/page3', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody(pov) self.assertHeader('Connection', 'Keep-Alive') # Remove the keep-alive header again. self.getPage('/page3') self.assertStatus('200 OK') self.assertBody(pov) # Apache, for example, may emit a Connection header even for HTTP/1.0 # self.assertNoHeader("Connection") class PipelineTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_HTTP11_Timeout(self): # If we timeout without sending any data, # the server will close the conn with a 408. if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Connect but send nothing. self.persistent = True conn = self.HTTP_CONN conn.auto_open = False conn.connect() # Wait for our socket timeout time.sleep(timeout * 2) # The request should have returned 408 already. response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 408) conn.close() # Connect but send half the headers only. self.persistent = True conn = self.HTTP_CONN conn.auto_open = False conn.connect() conn.send(b'GET /hello HTTP/1.1') conn.send(('Host: %s' % self.HOST).encode('ascii')) # Wait for our socket timeout time.sleep(timeout * 2) # The conn should have already sent 408. response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 408) conn.close() def test_HTTP11_Timeout_after_request(self): # If we timeout after at least one request has succeeded, # the server will close the conn without 408. if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Make an initial request self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/timeout?t=%s' % timeout, skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() self.assertBody(str(timeout)) # Make a second request on the same socket conn._output(b'GET /hello HTTP/1.1') conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() self.assertBody('Hello, world!') # Wait for our socket timeout time.sleep(timeout * 2) # Make another request on the same socket, which should error conn._output(b'GET /hello HTTP/1.1') conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() response = conn.response_class(conn.sock, method='GET') msg = ( "Writing to timed out socket didn't fail as it should have: %s") try: response.begin() except Exception: if not isinstance(sys.exc_info()[1], (socket.error, BadStatusLine)): self.fail(msg % sys.exc_info()[1]) else: if response.status != 408: self.fail(msg % response.read()) conn.close() # Make another request on a new socket, which should work self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() self.assertBody(pov) # Make another request on the same socket, # but timeout on the headers conn.send(b'GET /hello HTTP/1.1') # Wait for our socket timeout time.sleep(timeout * 2) response = conn.response_class(conn.sock, method='GET') try: response.begin() except Exception: if not isinstance(sys.exc_info()[1], (socket.error, BadStatusLine)): self.fail(msg % sys.exc_info()[1]) else: if response.status != 408: self.fail(msg % response.read()) conn.close() # Retry the request on a new connection, which should work self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() self.assertBody(pov) conn.close() def test_HTTP11_pipelining(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Test pipelining. httplib doesn't support this directly. self.persistent = True conn = self.HTTP_CONN # Put request 1 conn.putrequest('GET', '/hello', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() for trial in range(5): # Put next request conn._output(b'GET /hello HTTP/1.1') conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() # Retrieve previous response response = conn.response_class(conn.sock, method='GET') # there is a bug in python3 regarding the buffering of # ``conn.sock``. Until that bug get's fixed we will # monkey patch the ``response`` instance. # https://bugs.python.org/issue23377 response.fp = conn.sock.makefile('rb', 0) response.begin() body = response.read(13) self.assertEqual(response.status, 200) self.assertEqual(body, b'Hello, world!') # Retrieve final response response = conn.response_class(conn.sock, method='GET') response.begin() body = response.read() self.assertEqual(response.status, 200) self.assertEqual(body, b'Hello, world!') conn.close() def test_100_Continue(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' self.persistent = True conn = self.HTTP_CONN # Try a page without an Expect request header first. # Note that httplib's response.begin automatically ignores # 100 Continue responses, so we must manually check for it. try: conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Content-Type', 'text/plain') conn.putheader('Content-Length', '4') conn.endheaders() conn.send(ntob("d'oh")) response = conn.response_class(conn.sock, method='POST') version, status, reason = response._read_status() self.assertNotEqual(status, 100) finally: conn.close() # Now try a page with an Expect header... try: conn.connect() conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Content-Type', 'text/plain') conn.putheader('Content-Length', '17') conn.putheader('Expect', '100-continue') conn.endheaders() response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() self.assertEqual(status, 100) while True: line = response.fp.readline().strip() if line: self.fail( '100 Continue should not output any headers. Got %r' % line) else: break # ...send the body body = b'I am a small file' conn.send(body) # ...get the final response response.begin() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(200) self.assertBody("thanks for '%s'" % body) finally: conn.close() class ConnectionTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_readall_or_close(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' if self.scheme == 'https': self.HTTP_CONN = HTTPSConnection else: self.HTTP_CONN = HTTPConnection # Test a max of 0 (the default) and then reset to what it was above. old_max = cherrypy.server.max_request_body_size for new_max in (0, old_max): cherrypy.server.max_request_body_size = new_max self.persistent = True conn = self.HTTP_CONN # Get a POST page with an error conn.putrequest('POST', '/err_before_read', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Content-Type', 'text/plain') conn.putheader('Content-Length', '1000') conn.putheader('Expect', '100-continue') conn.endheaders() response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() self.assertEqual(status, 100) while True: skip = response.fp.readline().strip() if not skip: break # ...send the body conn.send(ntob('x' * 1000)) # ...get the final response response.begin() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(500) # Now try a working page with an Expect header... conn._output(b'POST /upload HTTP/1.1') conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._output(b'Content-Type: text/plain') conn._output(b'Content-Length: 17') conn._output(b'Expect: 100-continue') conn._send_output() response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() self.assertEqual(status, 100) while True: skip = response.fp.readline().strip() if not skip: break # ...send the body body = b'I am a small file' conn.send(body) # ...get the final response response.begin() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(200) self.assertBody("thanks for '%s'" % body) conn.close() def test_No_Message_Body(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Set our HTTP_CONN to an instance so it persists between requests. self.persistent = True # Make the first request and assert there's no "Connection: close". self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) self.assertNoHeader('Connection') # Make a 204 request on the same connection. self.getPage('/custom/204') self.assertStatus(204) self.assertNoHeader('Content-Length') self.assertBody('') self.assertNoHeader('Connection') # Make a 304 request on the same connection. self.getPage('/custom/304') self.assertStatus(304) self.assertNoHeader('Content-Length') self.assertBody('') self.assertNoHeader('Connection') def test_Chunked_Encoding(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() if (hasattr(self, 'harness') and 'modpython' in self.harness.__class__.__name__.lower()): # mod_python forbids chunked encoding return self.skip() self.PROTOCOL = 'HTTP/1.1' # Set our HTTP_CONN to an instance so it persists between requests. self.persistent = True conn = self.HTTP_CONN # Try a normal chunked request (with extensions) body = ntob('8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n' 'Content-Type: application/json\r\n' '\r\n') conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Transfer-Encoding', 'chunked') conn.putheader('Trailer', 'Content-Type') # Note that this is somewhat malformed: # we shouldn't be sending Content-Length. # RFC 2616 says the server should ignore it. conn.putheader('Content-Length', '3') conn.endheaders() conn.send(body) response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus('200 OK') self.assertBody("thanks for '%s'" % b'xx\r\nxxxxyyyyy') # Try a chunked request that exceeds server.max_request_body_size. # Note that the delimiters and trailer are included. body = ntob('3e3\r\n' + ('x' * 995) + '\r\n0\r\n\r\n') conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Transfer-Encoding', 'chunked') conn.putheader('Content-Type', 'text/plain') # Chunked requests don't need a content-length # # conn.putheader("Content-Length", len(body)) conn.endheaders() conn.send(body) response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(413) conn.close() def test_Content_Length_in(self): # Try a non-chunked request where Content-Length exceeds # server.max_request_body_size. Assert error before body send. self.persistent = True conn = self.HTTP_CONN conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Content-Type', 'text/plain') conn.putheader('Content-Length', '9999') conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(413) self.assertBody('The entity sent with the request exceeds ' 'the maximum allowed bytes.') conn.close() def test_Content_Length_out_preheaders(self): # Try a non-chunked response where Content-Length is less than # the actual bytes in the response body. self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/custom_cl?body=I+have+too+many+bytes&cl=5', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(500) self.assertBody( 'The requested resource returned more bytes than the ' 'declared Content-Length.') conn.close() def test_Content_Length_out_postheaders(self): # Try a non-chunked response where Content-Length is less than # the actual bytes in the response body. self.persistent = True conn = self.HTTP_CONN conn.putrequest( 'GET', '/custom_cl?body=I+too&body=+have+too+many&cl=5', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(200) self.assertBody('I too') conn.close() def test_598(self): tmpl = '{scheme}://{host}:{port}/one_megabyte_of_a/' url = tmpl.format( scheme=self.scheme, host=self.HOST, port=self.PORT, ) remote_data_conn = urllib.request.urlopen(url) buf = remote_data_conn.read(512) time.sleep(timeout * 0.6) remaining = (1024 * 1024) - 512 while remaining: data = remote_data_conn.read(remaining) if not data: break else: buf += data remaining -= len(data) self.assertEqual(len(buf), 1024 * 1024) self.assertEqual(buf, ntob('a' * 1024 * 1024)) self.assertEqual(remaining, 0) remote_data_conn.close() def setup_upload_server(): class Root: @cherrypy.expose def upload(self): if not cherrypy.request.method == 'POST': raise AssertionError("'POST' != request.method %r" % cherrypy.request.method) return "thanks for '%s'" % tonative(cherrypy.request.body.read()) cherrypy.tree.mount(Root()) cherrypy.config.update({ 'server.max_request_body_size': 1001, 'server.socket_timeout': 10, 'server.accepted_queue_size': 5, 'server.accepted_queue_timeout': 0.1, }) reset_names = 'ECONNRESET', 'WSAECONNRESET' socket_reset_errors = [ getattr(errno, name) for name in reset_names if hasattr(errno, name) ] 'reset error numbers available on this platform' socket_reset_errors += [ # Python 3.5 raises an http.client.RemoteDisconnected # with this message 'Remote end closed connection without response', ] class LimitedRequestQueueTests(helper.CPWebCase): setup_server = staticmethod(setup_upload_server) def test_queue_full(self): conns = [] overflow_conn = None try: # Make 15 initial requests and leave them open, which should use # all of wsgiserver's WorkerThreads and fill its Queue. for i in range(15): conn = self.HTTP_CONN(self.HOST, self.PORT) conn.putrequest('POST', '/upload', skip_host=True) conn.putheader('Host', self.HOST) conn.putheader('Content-Type', 'text/plain') conn.putheader('Content-Length', '4') conn.endheaders() conns.append(conn) # Now try a 16th conn, which should be closed by the # server immediately. overflow_conn = self.HTTP_CONN(self.HOST, self.PORT) # Manually connect since httplib won't let us set a timeout for res in socket.getaddrinfo(self.HOST, self.PORT, 0, socket.SOCK_STREAM): af, socktype, proto, canonname, sa = res overflow_conn.sock = socket.socket(af, socktype, proto) overflow_conn.sock.settimeout(5) overflow_conn.sock.connect(sa) break overflow_conn.putrequest('GET', '/', skip_host=True) overflow_conn.putheader('Host', self.HOST) overflow_conn.endheaders() response = overflow_conn.response_class( overflow_conn.sock, method='GET', ) try: response.begin() except socket.error as exc: if exc.args[0] in socket_reset_errors: pass # Expected. else: tmpl = ( 'Overflow conn did not get RST. ' 'Got {exc.args!r} instead' ) raise AssertionError(tmpl.format(**locals())) except BadStatusLine: # This is a special case in OS X. Linux and Windows will # RST correctly. assert sys.platform == 'darwin' else: raise AssertionError('Overflow conn did not get RST ') finally: for conn in conns: conn.send(b'done') response = conn.response_class(conn.sock, method='POST') response.begin() self.body = response.read() self.assertBody("thanks for 'done'") self.assertEqual(response.status, 200) conn.close() if overflow_conn: overflow_conn.close() class BadRequestTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_No_CRLF(self): self.persistent = True conn = self.HTTP_CONN conn.send(b'GET /hello HTTP/1.1\n\n') response = conn.response_class(conn.sock, method='GET') response.begin() self.body = response.read() self.assertBody('HTTP requires CRLF terminators') conn.close() conn.connect() conn.send(b'GET /hello HTTP/1.1\r\n\n') response = conn.response_class(conn.sock, method='GET') response.begin() self.body = response.read() self.assertBody('HTTP requires CRLF terminators') conn.close()