bazarr/libs/cherrypy/test/test_http.py

"""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