import os import pytest import mock import twisted.internet import twisted.test from pappyproxy import http from pappyproxy import macros from pappyproxy import config from pappyproxy.proxy import ProxyClient, ProxyClientFactory, ProxyServerFactory from testutil import mock_deferred, func_deleted, func_ignored_deferred, func_ignored, no_tcp from twisted.internet.protocol import ServerFactory from twisted.test.iosim import FakeTransport from twisted.internet import defer, reactor #################### ## Fixtures MANGLED_REQ = 'GET /mangled HTTP/1.1\r\n\r\n' MANGLED_RSP = 'HTTP/1.1 500 MANGLED\r\nContent-Length: 0\r\n\r\n' @pytest.fixture def unconnected_proxyserver(mocker): mocker.patch("twisted.test.iosim.FakeTransport.startTLS") mocker.patch("pappyproxy.proxy.load_certs_from_dir", new=mock_generate_cert) factory = ProxyServerFactory() protocol = factory.buildProtocol(('127.0.0.1', 0)) protocol.makeConnection(FakeTransport(protocol, True)) return protocol @pytest.fixture def proxyserver(mocker): mocker.patch("twisted.test.iosim.FakeTransport.startTLS") mocker.patch("pappyproxy.proxy.load_certs_from_dir", new=mock_generate_cert) factory = ProxyServerFactory() protocol = factory.buildProtocol(('127.0.0.1', 0)) protocol.makeConnection(FakeTransport(protocol, True)) protocol.lineReceived('CONNECT https://www.AAAA.BBBB:443 HTTP/1.1') protocol.lineReceived('') protocol.transport.getOutBuffer() return protocol @pytest.fixture def proxy_connection(): @defer.inlineCallbacks def gen_connection(send_data, new_req=False, new_rsp=False, drop_req=False, drop_rsp=False): factory = ProxyClientFactory(http.Request(send_data)) macro = gen_mangle_macro(new_req, new_rsp, drop_req, drop_rsp) factory.intercepting_macros['pappy_mangle'] = macro protocol = factory.buildProtocol(None) tr = FakeTransport(protocol, True) protocol.makeConnection(tr) sent = yield protocol.data_defer print sent defer.returnValue((protocol, sent, factory.data_defer)) return gen_connection @pytest.fixture def in_scope_true(mocker): new_in_scope = mock.MagicMock() new_in_scope.return_value = True mocker.patch("pappyproxy.context.in_scope", new=new_in_scope) return new_in_scope @pytest.fixture def in_scope_false(mocker): new_in_scope = mock.MagicMock() new_in_scope.return_value = False mocker.patch("pappyproxy.context.in_scope", new=new_in_scope) return new_in_scope ## Autorun fixtures @pytest.fixture(autouse=True) def ignore_save(mocker): mocker.patch("pappyproxy.http.Request.async_deep_save", func_ignored_deferred) #################### ## Mock functions def mock_generate_cert(cert_dir): private_key = ('-----BEGIN PRIVATE KEY-----\n' 'MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDAoClrYUEB7lM0\n' 'zQaKkXZVG2d1Bu9hV8urpx0gNXMbyZ2m3xb+sKZju/FHPuWenA4KaN5gRUT+oLfv\n' 'tnF6Ia0jpRNWnX0Fyn/irdg1BWGJn7k7mJ2D0NXZQczn2+xxY05599NfGWqNKCYy\n' 'jhSwPsUK+sGJqi7aSDdlS97ZTjrQVTTFsC0+kSu4lS5fsWXxqrKLa6Ao8W7abVRO\n' 'JHazh/cxM4UKpgWU+E6yD4o4ZgHY+SMTVUh/IOM8DuOVyLEWtx4oLNiLMlpWT3qy\n' '4IMpOF6VuU6JF2HGV13SoJfhsLXsPRbLVTAnZvJcZwtgDm6NfKapU8W8olkDV1Bf\n' 'YQEMSNX7AgMBAAECggEBAII0wUrAdrzjaIMsg9tu8FofKBPHGFDok9f4Iov/FUwX\n' 'QOXnrxeCOTb5d+L89SH9ws/ui0LwD+8+nJcA8DvqP6r0jtnhov0jIMcNVDSi6oeo\n' '3AEY7ICJzcQJ4oRn+K+8vPNdPhfuikPYe9l4iSuJgpAlaGWyD/GlFyz12DFz2/Wu\n' 'NIcqR1ucvezRHn3eGMtvDv2WGaN4ifUc30k8XgSUesmwSI6beb5+hxq7wXfsurnP\n' 'EUrPY9ts3lfiAgxzTKOuj1VR5hn7cJyLN8jF0mZs4D6eSSHorIddhmaNiCq5ZbMd\n' 'QdlDiPvnXHT41OoXOb7tDEt7SGoiRh2noCZ1aZiSziECgYEA+tuPPLYWU6JRB6EW\n' 'PhbcXQbh3vML7eT1q7DOz0jYCojgT2+k7EWSI8T830oQyjbpe3Z86XEgH7UBjUgq\n' '27nJ4E6dQDYGbYCKEklOoCGLE7A60i1feIz8otOQRrbQ4jcpibEgscA6gzHmunYf\n' 'De5euUgYW+Rq2Vmr6/NzUaUgui8CgYEAxJMDwPOGgiLM1cczlaSIU9Obz+cVnwWn\n' 'nsdKYMto2V3yKLydDfjsgOgzxHOxxy+5L645TPxK6CkiISuhJ93kAFFtx+1sCBCT\n' 'tVzY5robVAekxA9tlPIxtsn3+/axx3n6HnV0oA/XtxkuOS5JImgEdXqFwJZkerGE\n' 'waftIU2FCfUCgYEArl8+ErJzlJEIiCgWIPSdGuD00pfZW/TCPCT7rKRy3+fDHBR7\n' '7Gxzp/9+0utV/mnrJBH5w/8JmGCmgoF+oRtk01FyBzdGgolN8GYajD6kwPvH917o\n' 'tRAzcC9lY3IigoxbiEWid0wqoBVoz4XaEkH2gA44OG/vQcQOOEYSi9cfh6sCgYBg\n' 'KLaOXdJvuIxRCzgNvMW/k+VFh3pJJx//COg2f2qT4mQCT3nYiutOh8hDEoFluc+y\n' 'Jlz7bvNJrE14wnn8IYxWJ383bMoLC+jlsDyeaW3S5kZQbmehk/SDwTrg86W1udKD\n' 'sdtSLU3N0LCO4jh+bzm3Ki9hrXALoOkbPoU+ZEhvPQKBgQDf79XQ3RNxZSk+eFyq\n' 'qD8ytVqxEoD+smPDflXXseVH6o+pNWrF8+A0KqmO8c+8KVzWj/OfULO6UbKd3E+x\n' '4JGkWu9yF1lEgtHgibF2ER8zCSIL4ikOEasPCkrKj5SrS4Q+j4u5ha76dIc2CVu1\n' 'hkX2PQ1xU4ocu06k373sf73A4Q==\n' '-----END PRIVATE KEY-----') ca_key = ('-----BEGIN CERTIFICATE-----\n' 'MIIDjzCCAncCFQCjC8r+I4xa7JoGUJYGOTcqDROA0DANBgkqhkiG9w0BAQsFADBg\n' 'MQswCQYDVQQGEwJVUzERMA8GA1UECBMITWljaGlnYW4xEjAQBgNVBAcTCUFubiBB\n' 'cmJvcjEUMBIGA1UEChMLUGFwcHkgUHJveHkxFDASBgNVBAMTC1BhcHB5IFByb3h5\n' 'MB4XDTE1MTEyMDIxMTEzOVoXDTI1MTExNzIxMTEzOVowYDELMAkGA1UEBhMCVVMx\n' 'ETAPBgNVBAgTCE1pY2hpZ2FuMRIwEAYDVQQHEwlBbm4gQXJib3IxFDASBgNVBAoT\n' 'C1BhcHB5IFByb3h5MRQwEgYDVQQDEwtQYXBweSBQcm94eTCCASIwDQYJKoZIhvcN\n' 'AQEBBQADggEPADCCAQoCggEBAMCgKWthQQHuUzTNBoqRdlUbZ3UG72FXy6unHSA1\n' 'cxvJnabfFv6wpmO78Uc+5Z6cDgpo3mBFRP6gt++2cXohrSOlE1adfQXKf+Kt2DUF\n' 'YYmfuTuYnYPQ1dlBzOfb7HFjTnn3018Zao0oJjKOFLA+xQr6wYmqLtpIN2VL3tlO\n' 'OtBVNMWwLT6RK7iVLl+xZfGqsotroCjxbtptVE4kdrOH9zEzhQqmBZT4TrIPijhm\n' 'Adj5IxNVSH8g4zwO45XIsRa3Higs2IsyWlZPerLggyk4XpW5TokXYcZXXdKgl+Gw\n' 'tew9FstVMCdm8lxnC2AObo18pqlTxbyiWQNXUF9hAQxI1fsCAwEAAaNFMEMwEgYD\n' 'VR0TAQH/BAgwBgEB/wIBADAOBgNVHQ8BAf8EBAMCAQYwHQYDVR0OBBYEFNo5o+5e\n' 'a0sNMlW/75VgGJCv2AcJMA0GCSqGSIb3DQEBCwUAA4IBAQBdJDhxbmoEe27bD8me\n' 'YTcLGjs/StKkSil7rLbX+tBCwtkm5UEEejBuAhKk2FuAXW8yR1FqKJSZwVCAocBT\n' 'Bo/+97Ee+h7ywrRFhATEr9D/TbbHKOjCjDzOMl9yLZa2DKErZjbI30ZD6NafWS/X\n' 'hx5X1cGohHcVVzT4jIgUEU70vvYfNn8CTZm4oJ7qqRe/uQPUYy0rwvbd60oprtGg\n' 'jNv1H5R4ODHUMBXAI9H7ft9cWrd0fBQjxhoj8pvgJXEZ52flXSqQc7qHLg1wO/zC\n' 'RUgpTcNAb2qCssBKbj+c1vKEPRUJfw6UYb0s1462rQNc8BgZiKaNbwokFmkAnjUg\n' 'AvnX\n' '-----END CERTIFICATE-----') return (ca_key, private_key) def gen_mangle_macro(modified_req=None, modified_rsp=None, drop_req=False, drop_rsp=False): macro = mock.MagicMock() if modified_req or drop_req: macro.async_req = True macro.intercept_requests = True if drop_req: newreq = None else: newreq = http.Request(modified_req) macro.async_mangle_request.return_value = mock_deferred(newreq) else: macro.intercept_requests = False if modified_rsp or drop_rsp: macro.async_rsp = True macro.intercept_responses = True if drop_rsp: newrsp = None else: newrsp = http.Response(modified_rsp) macro.async_mangle_response.return_value = mock_deferred(newrsp) else: macro.intercept_responses = False return macro def notouch_mangle_req(request): d = mock_deferred(request) return d def notouch_mangle_rsp(request): d = mock_deferred(request.response) return d def req_mangler_change(request): req = http.Request('GET /mangled HTTP/1.1\r\n\r\n') d = mock_deferred(req) return d def rsp_mangler_change(request): rsp = http.Response('HTTP/1.1 500 MANGLED\r\n\r\n') d = mock_deferred(rsp) return d def req_mangler_drop(request): return mock_deferred(None) def rsp_mangler_drop(request): return mock_deferred(None) #################### ## Unit test tests def test_proxy_server_fixture(unconnected_proxyserver): unconnected_proxyserver.transport.write('hello') assert unconnected_proxyserver.transport.getOutBuffer() == 'hello' @pytest.inlineCallbacks def test_mock_deferreds(): d = mock_deferred('Hello!') r = yield d assert r == 'Hello!' def test_deleted(): with pytest.raises(NotImplementedError): reactor.connectTCP("www.google.com", "80", ServerFactory) with pytest.raises(NotImplementedError): reactor.connectSSL("www.google.com", "80", ServerFactory) #################### ## Proxy Server Tests def test_proxy_server_connect(unconnected_proxyserver, mocker, in_scope_true): mocker.patch("twisted.internet.reactor.connectSSL") unconnected_proxyserver.lineReceived('CONNECT https://www.dddddd.fff:433 HTTP/1.1') unconnected_proxyserver.lineReceived('') assert unconnected_proxyserver.transport.getOutBuffer() == 'HTTP/1.1 200 Connection established\r\n\r\n' assert unconnected_proxyserver._request_obj.is_ssl def test_proxy_server_basic(proxyserver, mocker, in_scope_true): mocker.patch("twisted.internet.reactor.connectSSL") mocker.patch('pappyproxy.proxy.ProxyServer.setRawMode') proxyserver.lineReceived('GET / HTTP/1.1') proxyserver.lineReceived('') assert proxyserver.setRawMode.called args, kwargs = twisted.internet.reactor.connectSSL.call_args assert args[0] == 'www.AAAA.BBBB' assert args[1] == 443 @pytest.inlineCallbacks def test_proxy_client_nomangle(mocker, proxy_connection, in_scope_true): # Make the connection (prot, sent, retreq_deferred) = \ yield proxy_connection('GET / HTTP/1.1\r\n\r\n', None, None) assert sent.full_request == 'GET / HTTP/1.1\r\n\r\n' prot.lineReceived('HTTP/1.1 200 OK') prot.lineReceived('Content-Length: 0') prot.lineReceived('') ret_req = yield retreq_deferred response = ret_req.response.full_response assert response == 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n' @pytest.inlineCallbacks def test_proxy_client_mangle_req(mocker, proxy_connection, in_scope_true): # Make the connection (prot, sent, retreq_deferred) = \ yield proxy_connection('GET / HTTP/1.1\r\n\r\n', MANGLED_REQ, None) assert sent.full_request == 'GET /mangled HTTP/1.1\r\n\r\n' @pytest.inlineCallbacks def test_proxy_client_mangle_rsp(mocker, proxy_connection, in_scope_true): # Make the connection (prot, sent, retreq_deferred) = \ yield proxy_connection('GET / HTTP/1.1\r\n\r\n', None, MANGLED_RSP) prot.lineReceived('HTTP/1.1 200 OK') prot.lineReceived('Content-Length: 0') prot.lineReceived('') req = yield retreq_deferred response = req.response.full_response assert response == 'HTTP/1.1 500 MANGLED\r\nContent-Length: 0\r\n\r\n' @pytest.inlineCallbacks def test_proxy_drop_req(mocker, proxy_connection, in_scope_true): (prot, sent, retreq_deferred) = \ yield proxy_connection('GET / HTTP/1.1\r\n\r\n', None, None, True, False) assert sent is None @pytest.inlineCallbacks def test_proxy_drop_rsp(mocker, proxy_connection, in_scope_true): (prot, sent, retreq_deferred) = \ yield proxy_connection('GET / HTTP/1.1\r\n\r\n', None, None, False, True) prot.lineReceived('HTTP/1.1 200 OK') prot.lineReceived('Content-Length: 0') prot.lineReceived('') retreq = yield retreq_deferred assert retreq.response is None @pytest.inlineCallbacks def test_proxy_client_360_noscope(mocker, proxy_connection, in_scope_false): # Make the connection (prot, sent, retreq_deferred) = yield proxy_connection('GET / HTTP/1.1\r\n\r\n') assert sent.full_request == 'GET / HTTP/1.1\r\n\r\n' prot.lineReceived('HTTP/1.1 200 OK') prot.lineReceived('Content-Length: 0') prot.lineReceived('') req = yield retreq_deferred assert req.response.full_response == 'HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n'