path: root/test/unit/common/test_http_protocol.py

# Copyright (c) 2010-2022 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from argparse import Namespace
from io import BytesIO
import mock
import types
import unittest
import eventlet.wsgi
import six
from swift.common import http_protocol, swob


class TestSwiftHttpProtocol(unittest.TestCase):
    def _proto_obj(self):
        # Make an object we can exercise... note the base class's __init__()
        # does a bunch of work, so we just new up an object like eventlet.wsgi
        # does.
        proto_class = http_protocol.SwiftHttpProtocol
        try:
            the_obj = types.InstanceType(proto_class)
        except AttributeError:
            the_obj = proto_class.__new__(proto_class)
        # Install some convenience mocks
        the_obj.server = Namespace(app=Namespace(logger=mock.Mock()),
                                   url_length_limit=777,
                                   log=mock.Mock())
        the_obj.send_error = mock.Mock()

        return the_obj

    def test_swift_http_protocol_log_request(self):
        proto_obj = self._proto_obj()
        self.assertEqual(None, proto_obj.log_request('ignored'))

    def test_swift_http_protocol_log_message(self):
        proto_obj = self._proto_obj()

        proto_obj.log_message('a%sc', 'b')
        self.assertEqual([mock.call.error('ERROR WSGI: a%sc', 'b')],
                         proto_obj.server.app.logger.mock_calls)

    def test_swift_http_protocol_log_message_no_logger(self):
        # If the app somehow had no logger attribute or it was None, don't blow
        # up
        proto_obj = self._proto_obj()
        delattr(proto_obj.server.app, 'logger')

        proto_obj.log_message('a%sc', 'b')
        self.assertEqual([mock.call.info('ERROR WSGI: a%sc', 'b')],
                         proto_obj.server.log.mock_calls)

        proto_obj.server.log.reset_mock()
        proto_obj.server.app.logger = None

        proto_obj.log_message('a%sc', 'b')
        self.assertEqual([mock.call.info('ERROR WSGI: a%sc', 'b')],
                         proto_obj.server.log.mock_calls)

    def test_swift_http_protocol_parse_request_no_proxy(self):
        proto_obj = self._proto_obj()
        proto_obj.raw_requestline = b'jimmy jam'
        proto_obj.client_address = ('a', '123')

        self.assertEqual(False, proto_obj.parse_request())

        self.assertEqual([
            mock.call(400, "Bad HTTP/0.9 request type ('jimmy')"),
        ], proto_obj.send_error.mock_calls)
        self.assertEqual(('a', '123'), proto_obj.client_address)

    def test_request_line_cleanup(self):
        def do_test(line_from_socket, expected_line=None):
            if expected_line is None:
                expected_line = line_from_socket

            proto_obj = self._proto_obj()
            proto_obj.raw_requestline = line_from_socket
            with mock.patch('swift.common.http_protocol.wsgi.HttpProtocol') \
                    as mock_super:
                proto_obj.parse_request()

            self.assertEqual([mock.call.parse_request(proto_obj)],
                             mock_super.mock_calls)
            self.assertEqual(proto_obj.raw_requestline, expected_line)

        do_test(b'GET / HTTP/1.1')
        do_test(b'GET /%FF HTTP/1.1')

        if not six.PY2:
            do_test(b'GET /\xff HTTP/1.1', b'GET /%FF HTTP/1.1')
            do_test(b'PUT /Here%20Is%20A%20SnowMan:\xe2\x98\x83 HTTP/1.0',
                    b'PUT /Here%20Is%20A%20SnowMan%3A%E2%98%83 HTTP/1.0')
            do_test(
                b'POST /?and%20it=fixes+params&'
                b'PALMTREE=\xf0%9f\x8c%b4 HTTP/1.1',
                b'POST /?and+it=fixes+params&PALMTREE=%F0%9F%8C%B4 HTTP/1.1')


class ProtocolTest(unittest.TestCase):
    def _run_bytes_through_protocol(self, bytes_from_client):
        rfile = BytesIO(bytes_from_client)
        wfile = BytesIO()

        # All this fakery is needed to make the WSGI server process one
        # connection, possibly with multiple requests, in the main
        # greenthread. It doesn't hurt correctness if the function is called
        # in a separate greenthread, but it makes using the debugger harder.
        class FakeGreenthread(object):
            def link(self, a_callable, *args):
                a_callable(self, *args)

        class FakePool(object):
            def spawn(self, a_callable, *args, **kwargs):
                a_callable(*args, **kwargs)
                return FakeGreenthread()

            def spawn_n(self, a_callable, *args, **kwargs):
                a_callable(*args, **kwargs)

            def waitall(self):
                pass

        addr = ('127.0.0.1', 8359)
        fake_tcp_socket = mock.Mock(
            setsockopt=lambda *a: None,
            makefile=lambda mode, bufsize: rfile if 'r' in mode else wfile,
            getsockname=lambda *a: addr
        )
        fake_listen_socket = mock.Mock(
            accept=mock.MagicMock(
                side_effect=[[fake_tcp_socket, addr],
                             # KeyboardInterrupt breaks the WSGI server out of
                             # its infinite accept-process-close loop.
                             KeyboardInterrupt]),
            getsockname=lambda *a: addr)
        del fake_listen_socket.do_handshake

        # If we let the WSGI server close rfile/wfile then we can't access
        # their contents any more.
        with mock.patch.object(wfile, 'close', lambda: None), \
                mock.patch.object(rfile, 'close', lambda: None):
            eventlet.wsgi.server(
                fake_listen_socket, self.app,
                protocol=self.protocol_class,
                custom_pool=FakePool(),
                log_output=False,  # quiet the test run
            )
        return wfile.getvalue()


class TestSwiftHttpProtocolSomeMore(ProtocolTest):
    protocol_class = http_protocol.SwiftHttpProtocol

    @staticmethod
    def app(env, start_response):
        start_response("200 OK", [])
        return [swob.wsgi_to_bytes(env['RAW_PATH_INFO'])]

    def test_simple(self):
        bytes_out = self._run_bytes_through_protocol((
            b"GET /someurl HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertEqual(lines[0], b"HTTP/1.1 200 OK")  # sanity check
        self.assertEqual(lines[-1], b'/someurl')

    def test_quoted(self):
        bytes_out = self._run_bytes_through_protocol((
            b"GET /some%fFpath%D8%AA HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertEqual(lines[0], b"HTTP/1.1 200 OK")  # sanity check
        self.assertEqual(lines[-1], b'/some%fFpath%D8%AA')

    def test_messy(self):
        bytes_out = self._run_bytes_through_protocol((
            b"GET /oh\xffboy%what$now%E2%80%bd HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertEqual(lines[-1], b'/oh\xffboy%what$now%E2%80%bd')


class TestProxyProtocol(ProtocolTest):
    protocol_class = http_protocol.SwiftHttpProxiedProtocol

    @staticmethod
    def app(env, start_response):
        start_response("200 OK", [])
        body = '\r\n'.join([
            'got addr: %s %s' % (
                env.get("REMOTE_ADDR", "<missing>"),
                env.get("REMOTE_PORT", "<missing>")),
            'on addr: %s %s' % (
                env.get("SERVER_ADDR", "<missing>"),
                env.get("SERVER_PORT", "<missing>")),
            'https is %s (scheme %s)' % (
                env.get("HTTPS", "<missing>"),
                env.get("wsgi.url_scheme", "<missing>")),
        ]) + '\r\n'
        return [body.encode("utf-8")]

    def test_request_with_proxy(self):
        bytes_out = self._run_bytes_through_protocol((
            b"PROXY TCP4 192.168.0.1 192.168.0.11 56423 4433\r\n"
            b"GET /someurl HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertEqual(lines[0], b"HTTP/1.1 200 OK")  # sanity check
        self.assertEqual(lines[-3:], [
            b"got addr: 192.168.0.1 56423",
            b"on addr: 192.168.0.11 4433",
            b"https is <missing> (scheme http)",
        ])

    def test_request_with_proxy_https(self):
        bytes_out = self._run_bytes_through_protocol((
            b"PROXY TCP4 192.168.0.1 192.168.0.11 56423 443\r\n"
            b"GET /someurl HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertEqual(lines[0], b"HTTP/1.1 200 OK")  # sanity check
        self.assertEqual(lines[-3:], [
            b"got addr: 192.168.0.1 56423",
            b"on addr: 192.168.0.11 443",
            b"https is on (scheme https)",
        ])

    def test_multiple_requests_with_proxy(self):
        bytes_out = self._run_bytes_through_protocol((
            b"PROXY TCP4 192.168.0.1 192.168.0.11 56423 443\r\n"
            b"GET /someurl HTTP/1.1\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
            b"GET /otherurl HTTP/1.1\r\n"
            b"User-Agent: something or other\r\n"
            b"Connection: close\r\n"
            b"\r\n"
        ))

        lines = bytes_out.split(b"\r\n")
        self.assertEqual(lines[0], b"HTTP/1.1 200 OK")  # sanity check

        # the address in the PROXY line is applied to every request
        addr_lines = [l for l in lines if l.startswith(b"got addr")]
        self.assertEqual(addr_lines, [b"got addr: 192.168.0.1 56423"] * 2)
        addr_lines = [l for l in lines if l.startswith(b"on addr")]
        self.assertEqual(addr_lines, [b"on addr: 192.168.0.11 443"] * 2)
        addr_lines = [l for l in lines if l.startswith(b"https is")]
        self.assertEqual(addr_lines, [b"https is on (scheme https)"] * 2)

    def test_missing_proxy_line(self):
        bytes_out = self._run_bytes_through_protocol((
            # whoops, no PROXY line here
            b"GET /someurl HTTP/1.0\r\n"
            b"User-Agent: something or other\r\n"
            b"\r\n"
        ))

        lines = [l for l in bytes_out.split(b"\r\n") if l]
        self.assertIn(b"400 Invalid PROXY line", lines[0])

    def test_malformed_proxy_lines(self):
        for bad_line in [b'PROXY jojo',
                         b'PROXYjojo a b c d e',
                         b'PROXY a b c d e',  # bad INET protocol and family
                         ]:
            bytes_out = self._run_bytes_through_protocol(bad_line)
            lines = [l for l in bytes_out.split(b"\r\n") if l]
            self.assertIn(b"400 Invalid PROXY line", lines[0])

    def test_unknown_client_addr(self):
        # For "UNKNOWN", the rest of the line before the CRLF may be omitted by
        # the sender, and the receiver must ignore anything presented before
        # the CRLF is found.
        for unknown_line in [b'PROXY UNKNOWN',  # mimimal valid unknown
                             b'PROXY UNKNOWNblahblah',  # also valid
                             b'PROXY UNKNOWN a b c d']:
            bytes_out = self._run_bytes_through_protocol((
                unknown_line + (b"\r\n"
                                b"GET /someurl HTTP/1.0\r\n"
                                b"User-Agent: something or other\r\n"
                                b"\r\n")
            ))
            lines = [l for l in bytes_out.split(b"\r\n") if l]
            self.assertIn(b"200 OK", lines[0])

    def test_address_and_environ(self):
        # Make an object we can exercise... note the base class's __init__()
        # does a bunch of work, so we just new up an object like eventlet.wsgi
        # does.
        dummy_env = {'OTHER_ENV_KEY': 'OTHER_ENV_VALUE'}
        mock_protocol = mock.Mock(get_environ=lambda s: dummy_env)
        patcher = mock.patch(
            'swift.common.http_protocol.SwiftHttpProtocol', mock_protocol
        )
        self.mock_super = patcher.start()
        self.addCleanup(patcher.stop)

        proto_class = http_protocol.SwiftHttpProxiedProtocol
        try:
            proxy_obj = types.InstanceType(proto_class)
        except AttributeError:
            proxy_obj = proto_class.__new__(proto_class)

        # Install some convenience mocks
        proxy_obj.server = Namespace(app=Namespace(logger=mock.Mock()),
                                     url_length_limit=777,
                                     log=mock.Mock())
        proxy_obj.send_error = mock.Mock()

        proxy_obj.rfile = BytesIO(
            b'PROXY TCP4 111.111.111.111 222.222.222.222 111 222'
        )

        assert proxy_obj.handle()

        self.assertEqual(proxy_obj.client_address, ('111.111.111.111', '111'))
        self.assertEqual(proxy_obj.proxy_address, ('222.222.222.222', '222'))
        expected_env = {
            'SERVER_PORT': '222',
            'SERVER_ADDR': '222.222.222.222',
            'OTHER_ENV_KEY': 'OTHER_ENV_VALUE'
        }
        self.assertEqual(proxy_obj.get_environ(), expected_env)