mirror
/
bazarr
mirror of https://github.com/morpheus65535/bazarr
1
0
Fork 0
Code Issues Releases Wiki Activity
bazarr/libs/cherrypy/test/test_http.py

261 lines
9.7 KiB
Python
Raw Blame History

"""Tests for managing HTTP issues (malformed requests, etc)."""
import errno
import mimetypes
import socket
import sys
from unittest import mock
import six
import cherrypy
from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob
from cherrypy.test import helper
def encode_multipart_formdata(files):
"""Return (content_type, body) ready for httplib.HTTP instance.
files: a sequence of (name, filename, value) tuples for multipart uploads.
"""
BOUNDARY = '________ThIs_Is_tHe_bouNdaRY_$'
L = []
for key, filename, value in files:
L.append('--' + BOUNDARY)
L.append('Content-Disposition: form-data; name="%s"; filename="%s"' %
(key, filename))
ct = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
L.append('Content-Type: %s' % ct)
L.append('')
L.append(value)
L.append('--' + BOUNDARY + '--')
L.append('')
body = '\r\n'.join(L)
content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
return content_type, body
class HTTPTests(helper.CPWebCase):
def make_connection(self):
if self.scheme == 'https':
return HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
else:
return HTTPConnection('%s:%s' % (self.interface(), self.PORT))
@staticmethod
def setup_server():
class Root:
@cherrypy.expose
def index(self, *args, **kwargs):
return 'Hello world!'
@cherrypy.expose
@cherrypy.config(**{'request.process_request_body': False})
def no_body(self, *args, **kwargs):
return 'Hello world!'
@cherrypy.expose
def post_multipart(self, file):
"""Return a summary ("a * 65536\nb * 65536") of the uploaded
file.
"""
contents = file.file.read()
summary = []
curchar = None
count = 0
for c in contents:
if c == curchar:
count += 1
else:
if count:
if six.PY3:
curchar = chr(curchar)
summary.append('%s * %d' % (curchar, count))
count = 1
curchar = c
if count:
if six.PY3:
curchar = chr(curchar)
summary.append('%s * %d' % (curchar, count))
return ', '.join(summary)
@cherrypy.expose
def post_filename(self, myfile):
'''Return the name of the file which was uploaded.'''
return myfile.filename
cherrypy.tree.mount(Root())
cherrypy.config.update({'server.max_request_body_size': 30000000})
def test_no_content_length(self):
# "The presence of a message-body in a request is signaled by the
# inclusion of a Content-Length or Transfer-Encoding header field in
# the request's message-headers."
#
# Send a message with neither header and no body. Even though
# the request is of method POST, this should be OK because we set
# request.process_request_body to False for our handler.
c = self.make_connection()
c.request('POST', '/no_body')
response = c.getresponse()
self.body = response.fp.read()
self.status = str(response.status)
self.assertStatus(200)
self.assertBody(ntob('Hello world!'))
# Now send a message that has no Content-Length, but does send a body.
# Verify that CP times out the socket and responds
# with 411 Length Required.
if self.scheme == 'https':
c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
else:
c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
# `_get_content_length` is needed for Python 3.6+
with mock.patch.object(c, '_get_content_length', lambda body, method: None, create=True):
# `_set_content_length` is needed for Python 2.7-3.5
with mock.patch.object(c, '_set_content_length', create=True):
c.request('POST', '/')
response = c.getresponse()
self.body = response.fp.read()
self.status = str(response.status)
self.assertStatus(411)
def test_post_multipart(self):
alphabet = 'abcdefghijklmnopqrstuvwxyz'
# generate file contents for a large post
contents = ''.join([c * 65536 for c in alphabet])
# encode as multipart form data
files = [('file', 'file.txt', contents)]
content_type, body = encode_multipart_formdata(files)
body = body.encode('Latin-1')
# post file
c = self.make_connection()
c.putrequest('POST', '/post_multipart')
c.putheader('Content-Type', content_type)
c.putheader('Content-Length', str(len(body)))
c.endheaders()
c.send(body)
response = c.getresponse()
self.body = response.fp.read()
self.status = str(response.status)
self.assertStatus(200)
self.assertBody(', '.join(['%s * 65536' % c for c in alphabet]))
def test_post_filename_with_special_characters(self):
'''Testing that we can handle filenames with special characters. This
was reported as a bug in:
https://github.com/cherrypy/cherrypy/issues/1146/
https://github.com/cherrypy/cherrypy/issues/1397'''
# We'll upload a bunch of files with differing names.
fnames = ['boop.csv', 'foo, bar.csv', 'bar, xxxx.csv', 'file"name.csv',
'file;name.csv', 'file; name.csv']
for fname in fnames:
files = [('myfile', fname, 'yunyeenyunyue')]
content_type, body = encode_multipart_formdata(files)
body = body.encode('Latin-1')
# post file
c = self.make_connection()
c.putrequest('POST', '/post_filename')
c.putheader('Content-Type', content_type)
c.putheader('Content-Length', str(len(body)))
c.endheaders()
c.send(body)
response = c.getresponse()
self.body = response.fp.read()
self.status = str(response.status)
self.assertStatus(200)
self.assertBody(fname)
def test_malformed_request_line(self):
if getattr(cherrypy.server, 'using_apache', False):
return self.skip('skipped due to known Apache differences...')
# Test missing version in Request-Line
c = self.make_connection()
c._output(ntob('GET /'))
c._send_output()
if hasattr(c, 'strict'):
response = c.response_class(c.sock, strict=c.strict, method='GET')
else:
# Python 3.2 removed the 'strict' feature, saying:
# "http.client now always assumes HTTP/1.x compliant servers."
response = c.response_class(c.sock, method='GET')
response.begin()
self.assertEqual(response.status, 400)
self.assertEqual(response.fp.read(22), ntob('Malformed Request-Line'))
c.close()
def test_request_line_split_issue_1220(self):
Request_URI = (
'/index?intervenant-entreprise-evenement_classaction=evenement-mailremerciements'
'&_path=intervenant-entreprise-evenement&intervenant-entreprise-evenement_action-id=19404'
'&intervenant-entreprise-evenement_id=19404&intervenant-entreprise_id=28092'
)
self.assertEqual(len('GET %s HTTP/1.1\r\n' % Request_URI), 256)
self.getPage(Request_URI)
self.assertBody('Hello world!')
def test_malformed_header(self):
c = self.make_connection()
c.putrequest('GET', '/')
c.putheader('Content-Type', 'text/plain')
# See https://github.com/cherrypy/cherrypy/issues/941
c._output(ntob('Re, 1.2.3.4#015#012'))
c.endheaders()
response = c.getresponse()
self.status = str(response.status)
self.assertStatus(400)
self.body = response.fp.read(20)
self.assertBody('Illegal header line.')
def test_http_over_https(self):
if self.scheme != 'https':
return self.skip('skipped (not running HTTPS)... ')
# Try connecting without SSL.
conn = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
conn.putrequest('GET', '/', skip_host=True)
conn.putheader('Host', self.HOST)
conn.endheaders()
response = conn.response_class(conn.sock, method='GET')
try:
response.begin()
self.assertEqual(response.status, 400)
self.body = response.read()
self.assertBody('The client sent a plain HTTP request, but this '
'server only speaks HTTPS on this port.')
except socket.error:
e = sys.exc_info()[1]
# "Connection reset by peer" is also acceptable.
if e.errno != errno.ECONNRESET:
raise
def test_garbage_in(self):
# Connect without SSL regardless of server.scheme
c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
c._output(ntob('gjkgjklsgjklsgjkljklsg'))
c._send_output()
response = c.response_class(c.sock, method='GET')
try:
response.begin()
self.assertEqual(response.status, 400)
self.assertEqual(response.fp.read(22),
ntob('Malformed Request-Line'))
c.close()
except socket.error:
e = sys.exc_info()[1]
# "Connection reset by peer" is also acceptable.
if e.errno != errno.ECONNRESET:
raise
Reference in New Issue View Git Blame Copy Permalink