Release 0.0.2
Soooo much stuff. Features, bugfixes, all that.
This commit is contained in:
Rob Glew
parent
b9692b451e
commit
f4bbd15c68
40 changed files with 6916 additions and 1209 deletions
|
|
@ -56,21 +56,18 @@ def test_gen_filter_by_all_request():
|
|||
# Cookie key
|
||||
r = Request('GET / HTTP/1.1\r\n')
|
||||
r.cookies['hello'] = 'world'
|
||||
r.update_from_objects()
|
||||
assert f(r)
|
||||
assert not fn(r)
|
||||
|
||||
# Cookie value
|
||||
r = Request('GET / HTTP/1.1\r\n')
|
||||
r.cookies['world'] = 'hello'
|
||||
r.update_from_objects()
|
||||
assert f(r)
|
||||
assert not fn(r)
|
||||
|
||||
# Nowhere in cookie
|
||||
r = Request('GET / HTTP/1.1\r\n')
|
||||
r.cookies['world'] = 'sucks'
|
||||
r.update_from_objects()
|
||||
assert not f(r)
|
||||
assert fn(r)
|
||||
|
||||
|
|
@ -123,7 +120,6 @@ def test_gen_filter_by_all_response(http_request):
|
|||
r = Response('HTTP/1.1 200 OK\r\n')
|
||||
http_request.response = r
|
||||
r.add_cookie(ResponseCookie('hello=goodbye'))
|
||||
r.update_from_objects()
|
||||
assert f(http_request)
|
||||
assert not fn(http_request)
|
||||
|
||||
|
|
@ -131,7 +127,6 @@ def test_gen_filter_by_all_response(http_request):
|
|||
r = Response('HTTP/1.1 200 OK\r\n')
|
||||
http_request.response = r
|
||||
r.add_cookie(ResponseCookie('goodbye=hello'))
|
||||
r.update_from_objects()
|
||||
assert f(http_request)
|
||||
assert not fn(http_request)
|
||||
|
||||
|
|
@ -139,7 +134,6 @@ def test_gen_filter_by_all_response(http_request):
|
|||
r = Response('HTTP/1.1 200 OK\r\n')
|
||||
http_request.response = r
|
||||
r.add_cookie(ResponseCookie('goodbye=for real'))
|
||||
r.update_from_objects()
|
||||
assert not f(http_request)
|
||||
assert fn(http_request)
|
||||
|
||||
|
|
@ -153,7 +147,6 @@ def test_filter_by_host(http_request):
|
|||
assert fn(http_request)
|
||||
|
||||
http_request.headers['Host'] = 'vim.sexy'
|
||||
http_request.update_from_text()
|
||||
assert http_request.host == 'vim.sexy'
|
||||
assert f(http_request)
|
||||
assert not fn(http_request)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import base64
|
||||
import copy
|
||||
import gzip
|
||||
import json
|
||||
import pytest
|
||||
|
|
@ -6,6 +7,7 @@ import StringIO
|
|||
import zlib
|
||||
|
||||
from pappyproxy.pappy import http
|
||||
from pappyproxy.util import PappyException
|
||||
|
||||
####################
|
||||
# Helper Functions
|
||||
|
|
@ -201,7 +203,7 @@ def test_length_overflow():
|
|||
# Test throwing an exception when adding data after complete
|
||||
l = http.LengthData(100)
|
||||
l.add_data('A'*100)
|
||||
with pytest.raises(http.DataAlreadyComplete):
|
||||
with pytest.raises(PappyException):
|
||||
l.add_data('A')
|
||||
|
||||
def test_repeatable_dict_simple():
|
||||
|
|
@ -391,7 +393,6 @@ def test_request_simple():
|
|||
assert r.complete
|
||||
assert r.fragment == None
|
||||
assert r.full_request == headers+data
|
||||
assert r.header_len == len(headers)
|
||||
assert r.headers_complete
|
||||
assert r.host == 'www.test.com'
|
||||
assert r.is_ssl == False
|
||||
|
|
@ -424,8 +425,8 @@ def test_request_urlparams():
|
|||
def test(r):
|
||||
assert r.complete
|
||||
assert r.fragment == 'frag'
|
||||
assert r.get_params['p1'] == 'foo'
|
||||
assert r.get_params['p2'] == 'bar'
|
||||
assert r.url_params['p1'] == 'foo'
|
||||
assert r.url_params['p2'] == 'bar'
|
||||
assert r.full_request == ('GET /?p1=foo&p2=bar#frag HTTP/1.1\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'\r\n')
|
||||
|
|
@ -444,8 +445,8 @@ def test_request_questionmark_url():
|
|||
def test(r):
|
||||
assert r.complete
|
||||
assert r.fragment == 'frag'
|
||||
assert r.get_params['?/to/?p1'] == 'foo'
|
||||
assert r.get_params['p2'] == 'bar'
|
||||
assert r.url_params['?/to/?p1'] == 'foo'
|
||||
assert r.url_params['p2'] == 'bar'
|
||||
assert r.full_request == ('GET /path/??/to/?p1=foo&p2=bar#frag HTTP/1.1\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'\r\n')
|
||||
|
|
@ -472,6 +473,22 @@ def test_request_postparams():
|
|||
test(ru)
|
||||
test(rj)
|
||||
|
||||
def test_post_params_update():
|
||||
r = http.Request(('GET / HTTP/1.1\r\n'
|
||||
'Content-Type: application/x-www-form-urlencoded\r\n'
|
||||
'Content-Length: 7\r\n\r\n'
|
||||
'a=b&c=d'))
|
||||
r.post_params['c'] = 'e'
|
||||
assert r.full_request == ('GET / HTTP/1.1\r\n'
|
||||
'Content-Type: application/x-www-form-urlencoded\r\n'
|
||||
'Content-Length: 7\r\n\r\n'
|
||||
'a=b&c=e')
|
||||
r.post_params['a'] = 'f'
|
||||
assert r.full_request == ('GET / HTTP/1.1\r\n'
|
||||
'Content-Type: application/x-www-form-urlencoded\r\n'
|
||||
'Content-Length: 7\r\n\r\n'
|
||||
'a=f&c=e')
|
||||
|
||||
def test_headers_end():
|
||||
header_lines = [
|
||||
'GET / HTTP/1.1',
|
||||
|
|
@ -632,20 +649,22 @@ def test_request_to_json():
|
|||
r = http.Request()
|
||||
r.status_line = 'GET / HTTP/1.1'
|
||||
r.headers['content-length'] = 500
|
||||
r.tags = ['foo', 'bar']
|
||||
r.raw_data = 'AAAA'
|
||||
r.reqid = 1
|
||||
r.reqid = '1'
|
||||
|
||||
rsp = http.Response()
|
||||
rsp.status_line = 'HTTP/1.1 200 OK'
|
||||
rsp.rspid = 2
|
||||
rsp.rspid = '2'
|
||||
|
||||
r.response = rsp
|
||||
|
||||
expected_reqdata = {u'full_request': unicode(base64.b64encode(r.full_request)),
|
||||
u'response_id': rsp.rspid,
|
||||
u'response_id': str(rsp.rspid),
|
||||
u'port': 80,
|
||||
u'is_ssl': False,
|
||||
u'reqid': r.reqid,
|
||||
u'tags': ['foo', 'bar'],
|
||||
u'reqid': str(r.reqid),
|
||||
}
|
||||
|
||||
assert json.loads(r.to_json()) == expected_reqdata
|
||||
|
|
@ -659,7 +678,7 @@ def test_request_update_content_length():
|
|||
'Content-Length: 10\r\n\r\n'
|
||||
'AAAAAAAAAA'))
|
||||
|
||||
def test_request_blank_get_params():
|
||||
def test_request_blank_url_params():
|
||||
r = http.Request()
|
||||
r.add_line('GET /this/??-asdf/ HTTP/1.1')
|
||||
assert r.full_request == ('GET /this/??-asdf/ HTTP/1.1\r\n\r\n')
|
||||
|
|
@ -667,9 +686,9 @@ def test_request_blank_get_params():
|
|||
r = http.Request()
|
||||
r.add_line('GET /this/??-asdf/?a=b&c&d=ef HTTP/1.1')
|
||||
assert r.full_request == ('GET /this/??-asdf/?a=b&c&d=ef HTTP/1.1\r\n\r\n')
|
||||
assert r.get_params['?-asdf/?a'] == 'b'
|
||||
assert r.get_params['c'] == None
|
||||
assert r.get_params['d'] == 'ef'
|
||||
assert r.url_params['?-asdf/?a'] == 'b'
|
||||
assert r.url_params['c'] == None
|
||||
assert r.url_params['d'] == 'ef'
|
||||
|
||||
def test_request_blank():
|
||||
r = http.Request('\r\n\n\n')
|
||||
|
|
@ -726,7 +745,7 @@ def test_request_set_url():
|
|||
def test_request_set_url_params():
|
||||
r = http.Request('GET / HTTP/1.1\r\n')
|
||||
r.url = 'www.AAAA.BBBB?a=b&c=d#foo'
|
||||
assert r.get_params.all_pairs() == [('a','b'), ('c','d')]
|
||||
assert r.url_params.all_pairs() == [('a','b'), ('c','d')]
|
||||
assert r.fragment == 'foo'
|
||||
assert r.url == 'http://www.AAAA.BBBB?a=b&c=d#foo'
|
||||
r.port = 400
|
||||
|
|
@ -734,6 +753,24 @@ def test_request_set_url_params():
|
|||
r.is_ssl = True
|
||||
assert r.url == 'https://www.AAAA.BBBB:400?a=b&c=d#foo'
|
||||
|
||||
def test_request_copy():
|
||||
r = http.Request(('GET / HTTP/1.1\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA'))
|
||||
r2 = copy.copy(r)
|
||||
assert r2.full_request == ('GET / HTTP/1.1\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA')
|
||||
|
||||
def test_request_url_blankpath():
|
||||
r = http.Request()
|
||||
r.status_line = 'GET / HTTP/1.1'
|
||||
r.url = 'https://www.google.com'
|
||||
r.headers['Host'] = r.host
|
||||
r.url_params.from_dict({'foo': 'bar'})
|
||||
assert r.full_path == '/?foo=bar'
|
||||
assert r.url == 'https://www.google.com?foo=bar'
|
||||
|
||||
|
||||
####################
|
||||
## Response tests
|
||||
|
|
@ -749,11 +786,9 @@ def test_response_simple():
|
|||
'',
|
||||
]
|
||||
data = 'A'*100
|
||||
header_len = len('\r\n'.join(header_lines)+'\r\n')
|
||||
rf, rl, ru, rj = rsp_by_lines_and_full(header_lines, data)
|
||||
def test(r):
|
||||
assert r.complete
|
||||
assert r.header_len == header_len
|
||||
assert r.raw_data == data
|
||||
assert r.response_code == 200
|
||||
assert r.response_text == 'OK'
|
||||
|
|
@ -1100,3 +1135,55 @@ def test_response_newlines():
|
|||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA')
|
||||
|
||||
def test_copy_response():
|
||||
r = http.Response(('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA'))
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA')
|
||||
|
||||
r2 = copy.copy(r)
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'AAAA')
|
||||
|
||||
def test_response_add_cookie():
|
||||
r = http.Response(('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=bar\r\n\r\n'))
|
||||
r.add_cookie(http.ResponseCookie('foo=baz'))
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=bar\r\n'
|
||||
'Set-Cookie: foo=baz\r\n\r\n')
|
||||
|
||||
def test_response_set_cookie():
|
||||
r = http.Response(('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'))
|
||||
r.set_cookie(http.ResponseCookie('foo=bar'))
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=bar\r\n\r\n')
|
||||
|
||||
r.set_cookie(http.ResponseCookie('foo=baz'))
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=baz\r\n\r\n')
|
||||
|
||||
def test_response_delete_cookie():
|
||||
r = http.Response(('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=bar\r\n\r\n'))
|
||||
r.delete_cookie('foo')
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n\r\n')
|
||||
|
||||
r = http.Response(('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n'
|
||||
'Set-Cookie: foo=bar\r\n'
|
||||
'Set-Cookie: foo=baz\r\n\r\n'))
|
||||
r.delete_cookie('foo')
|
||||
assert r.full_response == ('HTTP/1.1 200 OK\r\n'
|
||||
'Content-Length: 0\r\n\r\n')
|
||||
|
|
|
|||
209
pappyproxy/tests/test_mangle.py
Normal file
209
pappyproxy/tests/test_mangle.py
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
import pytest
|
||||
import mock
|
||||
import pappyproxy
|
||||
|
||||
from pappyproxy.mangle import async_mangle_request, async_mangle_response
|
||||
from pappyproxy.http import Request, Response
|
||||
from testutil import no_tcp, no_database, func_deleted, mock_deferred, mock_deep_save, fake_saving
|
||||
|
||||
def retf(r):
|
||||
return False
|
||||
|
||||
@pytest.fixture
|
||||
def ignore_edit(mocker):
|
||||
new_edit = mock.MagicMock()
|
||||
new_edit.return_value = mock_deferred(None)
|
||||
mocker.patch('pappyproxy.console.edit_file', new=new_edit)
|
||||
|
||||
@pytest.fixture
|
||||
def ignore_delete(mocker):
|
||||
new_os_remove = mock.MagicMock()
|
||||
mocker.patch('os.remove', new=new_os_remove)
|
||||
return new_os_remove
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_logging(mocker):
|
||||
mocker.patch('pappyproxy.proxy.log')
|
||||
|
||||
@pytest.fixture
|
||||
def req():
|
||||
r = Request()
|
||||
r.status_line = 'GET / HTTP/1.1'
|
||||
r.host = 'www.ffffff.eeeeee'
|
||||
r.raw_data = 'AAAA'
|
||||
return r
|
||||
|
||||
@pytest.fixture
|
||||
def req_w_rsp(req):
|
||||
r = Response()
|
||||
r.status_line = 'HTTP/1.1 200 OK'
|
||||
r.headers['Test-Header'] = 'ABC123'
|
||||
r.raw_data = 'AAAA'
|
||||
req.response = r
|
||||
return req
|
||||
|
||||
@pytest.fixture
|
||||
def mock_tempfile(mocker):
|
||||
new_tfile_obj = mock.MagicMock()
|
||||
tfile_instance = mock.MagicMock()
|
||||
new_tfile_obj.return_value.__enter__.return_value = tfile_instance
|
||||
|
||||
tfile_instance.name = 'mockTemporaryFile'
|
||||
mocker.patch('tempfile.NamedTemporaryFile', new=new_tfile_obj)
|
||||
|
||||
new_open = mock.MagicMock()
|
||||
fake_file = mock.MagicMock(spec=file)
|
||||
new_open.return_value.__enter__.return_value = fake_file
|
||||
mocker.patch('__builtin__.open', new_open)
|
||||
|
||||
return (new_tfile_obj, tfile_instance, new_open, fake_file)
|
||||
|
||||
|
||||
########################
|
||||
## Test request mangling
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_request_edit(req, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req
|
||||
new_contents = ('GET / HTTP/1.1\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'BBBB')
|
||||
fake_file.read.return_value = new_contents
|
||||
new_req = yield async_mangle_request(r)
|
||||
assert not mock_deep_save.called
|
||||
assert tfile_obj.called
|
||||
assert tfile_instance.write.called
|
||||
assert tfile_instance.write.call_args == ((r.full_request,),)
|
||||
assert new_open.called
|
||||
assert fake_file.read.called
|
||||
|
||||
assert new_req.full_request == new_contents
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_request_edit_newlines(req, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req
|
||||
new_contents = ('GET / HTTP/1.1\r\n'
|
||||
'Test-Head: FOOBIE\n'
|
||||
'Content-Length: 4\n\r\n'
|
||||
'BBBB')
|
||||
fake_file.read.return_value = new_contents
|
||||
new_req = yield async_mangle_request(r)
|
||||
|
||||
assert new_req.full_request == ('GET / HTTP/1.1\r\n'
|
||||
'Test-Head: FOOBIE\r\n'
|
||||
'Content-Length: 4\r\n\r\n'
|
||||
'BBBB')
|
||||
assert new_req.headers['Test-Head'] == 'FOOBIE'
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_request_drop(req, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req
|
||||
new_contents = ''
|
||||
fake_file.read.return_value = new_contents
|
||||
new_req = yield async_mangle_request(r)
|
||||
|
||||
assert new_req is None
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_request_edit_len(req, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req
|
||||
new_contents = ('GET / HTTP/1.1\r\n'
|
||||
'Test-Head: FOOBIE\n'
|
||||
'Content-Length: 4\n\r\n'
|
||||
'BBBBAAAA')
|
||||
fake_file.read.return_value = new_contents
|
||||
new_req = yield async_mangle_request(r)
|
||||
|
||||
assert new_req.full_request == ('GET / HTTP/1.1\r\n'
|
||||
'Test-Head: FOOBIE\r\n'
|
||||
'Content-Length: 8\r\n\r\n'
|
||||
'BBBBAAAA')
|
||||
|
||||
|
||||
#########################
|
||||
## Test response mangling
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_response_edit(req_w_rsp, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is on, edit
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req_w_rsp
|
||||
old_rsp = r.response.full_response
|
||||
new_contents = ('HTTP/1.1 403 NOTOKIEDOKIE\r\n'
|
||||
'Content-Length: 4\r\n'
|
||||
'Other-Header: foobles\r\n\r\n'
|
||||
'BBBB')
|
||||
fake_file.read.return_value = new_contents
|
||||
mangled_rsp = yield async_mangle_response(r)
|
||||
assert not mock_deep_save.called
|
||||
assert tfile_obj.called
|
||||
assert tfile_instance.write.called
|
||||
assert tfile_instance.write.call_args == ((old_rsp,),)
|
||||
assert new_open.called
|
||||
assert fake_file.read.called
|
||||
|
||||
assert mangled_rsp.full_response == new_contents
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_response_newlines(req_w_rsp, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req_w_rsp
|
||||
old_rsp = r.response.full_response
|
||||
new_contents = ('HTTP/1.1 403 NOTOKIEDOKIE\n'
|
||||
'Content-Length: 4\n'
|
||||
'Other-Header: foobles\r\n\n'
|
||||
'BBBB')
|
||||
fake_file.read.return_value = new_contents
|
||||
mangled_rsp = yield async_mangle_response(r)
|
||||
|
||||
assert mangled_rsp.full_response == ('HTTP/1.1 403 NOTOKIEDOKIE\r\n'
|
||||
'Content-Length: 4\r\n'
|
||||
'Other-Header: foobles\r\n\r\n'
|
||||
'BBBB')
|
||||
assert mangled_rsp.headers['Other-Header'] == 'foobles'
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_response_drop(req_w_rsp, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req_w_rsp
|
||||
old_rsp = r.response.full_response
|
||||
new_contents = ''
|
||||
fake_file.read.return_value = new_contents
|
||||
mangled_rsp = yield async_mangle_response(r)
|
||||
|
||||
assert mangled_rsp is None
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_mangle_response_new_len(req_w_rsp, mock_deep_save, mock_tempfile,
|
||||
ignore_edit, ignore_delete):
|
||||
# Intercepting is off, request in scope
|
||||
tfile_obj, tfile_instance, new_open, fake_file = mock_tempfile
|
||||
r = req_w_rsp
|
||||
old_rsp = r.response.full_response
|
||||
new_contents = ('HTTP/1.1 403 NOTOKIEDOKIE\n'
|
||||
'Content-Length: 4\n'
|
||||
'Other-Header: foobles\r\n\n'
|
||||
'BBBBAAAA')
|
||||
fake_file.read.return_value = new_contents
|
||||
mangled_rsp = yield async_mangle_response(r)
|
||||
|
||||
assert mangled_rsp.full_response == ('HTTP/1.1 403 NOTOKIEDOKIE\r\n'
|
||||
'Content-Length: 8\r\n'
|
||||
'Other-Header: foobles\r\n\r\n'
|
||||
'BBBBAAAA')
|
||||
|
|
@ -1,11 +1,14 @@
|
|||
import os
|
||||
import pytest
|
||||
import mock
|
||||
import twisted.internet
|
||||
import twisted.test
|
||||
|
||||
from pappyproxy import http
|
||||
from pappyproxy import macros
|
||||
from pappyproxy import mangle
|
||||
from pappyproxy.proxy import ProxyClient, ProxyClientFactory, ProxyServer
|
||||
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
|
||||
|
|
@ -14,12 +17,14 @@ 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\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 = ServerFactory()
|
||||
factory.protocol = ProxyServer
|
||||
factory = ProxyServerFactory()
|
||||
protocol = factory.buildProtocol(('127.0.0.1', 0))
|
||||
protocol.makeConnection(FakeTransport(protocol, True))
|
||||
return protocol
|
||||
|
|
@ -28,8 +33,7 @@ def unconnected_proxyserver(mocker):
|
|||
def proxyserver(mocker):
|
||||
mocker.patch("twisted.test.iosim.FakeTransport.startTLS")
|
||||
mocker.patch("pappyproxy.proxy.load_certs_from_dir", new=mock_generate_cert)
|
||||
factory = ServerFactory()
|
||||
factory.protocol = ProxyServer
|
||||
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')
|
||||
|
|
@ -40,26 +44,40 @@ def proxyserver(mocker):
|
|||
@pytest.fixture
|
||||
def proxy_connection():
|
||||
@defer.inlineCallbacks
|
||||
def gen_connection(send_data):
|
||||
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 no_mangle(mocker):
|
||||
# # Don't call anything in mangle.py
|
||||
# mocker.patch("mangle.mangle_request", notouch_mangle_req)
|
||||
# mocker.patch("mangle.mangle_response", notouch_mangle_rsp)
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def ignore_save(mocker):
|
||||
mocker.patch("pappyproxy.http.Request.deep_save", func_ignored_deferred)
|
||||
mocker.patch("pappyproxy.http.Request.async_deep_save", func_ignored_deferred)
|
||||
|
||||
####################
|
||||
## Mock functions
|
||||
|
|
@ -117,32 +135,56 @@ def mock_generate_cert(cert_dir):
|
|||
'-----END CERTIFICATE-----')
|
||||
return (ca_key, private_key)
|
||||
|
||||
def notouch_mangle_req(request, conn_id):
|
||||
orig_req = http.Request(request.full_request)
|
||||
orig_req.port = request.port
|
||||
orig_req.is_ssl = request.is_ssl
|
||||
d = mock_deferred(orig_req)
|
||||
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.do_req = True
|
||||
if drop_req:
|
||||
newreq = None
|
||||
else:
|
||||
newreq = http.Request(modified_req)
|
||||
macro.async_mangle_request.return_value = mock_deferred(newreq)
|
||||
else:
|
||||
macro.do_req = False
|
||||
|
||||
if modified_rsp or drop_rsp:
|
||||
macro.async_rsp = True
|
||||
macro.do_rsp = True
|
||||
if drop_rsp:
|
||||
newrsp = None
|
||||
else:
|
||||
newrsp = http.Response(modified_rsp)
|
||||
macro.async_mangle_response.return_value = mock_deferred(newrsp)
|
||||
else:
|
||||
macro.do_rsp = False
|
||||
return macro
|
||||
|
||||
def notouch_mangle_req(request):
|
||||
d = mock_deferred(request)
|
||||
return d
|
||||
|
||||
def notouch_mangle_rsp(response, conn_id):
|
||||
req = http.Request()
|
||||
orig_rsp = http.Response(response.full_response)
|
||||
req.response = orig_rsp
|
||||
d = mock_deferred(req)
|
||||
def notouch_mangle_rsp(request):
|
||||
d = mock_deferred(request.response)
|
||||
return d
|
||||
|
||||
def req_mangler_change(request, conn_id):
|
||||
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, conn_id):
|
||||
req = http.Request()
|
||||
def rsp_mangler_change(request):
|
||||
rsp = http.Response('HTTP/1.1 500 MANGLED\r\n\r\n')
|
||||
req.response = rsp
|
||||
d = mock_deferred(req)
|
||||
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
|
||||
|
||||
|
|
@ -165,14 +207,14 @@ def test_deleted():
|
|||
####################
|
||||
## Proxy Server Tests
|
||||
|
||||
def test_proxy_server_connect(unconnected_proxyserver, mocker):
|
||||
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):
|
||||
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')
|
||||
|
|
@ -184,37 +226,60 @@ def test_proxy_server_basic(proxyserver, mocker):
|
|||
assert args[1] == 443
|
||||
|
||||
@pytest.inlineCallbacks
|
||||
def test_proxy_client_basic(mocker, proxy_connection):
|
||||
mocker.patch('pappyproxy.mangle.mangle_request', new=notouch_mangle_req)
|
||||
mocker.patch('pappyproxy.mangle.mangle_response', new=notouch_mangle_rsp)
|
||||
def test_proxy_client_nomangle(mocker, proxy_connection, in_scope_true):
|
||||
# Make the connection
|
||||
(prot, sent, resp_deferred) = yield proxy_connection('GET / HTTP/1.1\r\n\r\n')
|
||||
assert sent == 'GET / HTTP/1.1\r\n\r\n'
|
||||
(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 resp_deferred
|
||||
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):
|
||||
mocker.patch('pappyproxy.mangle.mangle_request', new=req_mangler_change)
|
||||
mocker.patch('pappyproxy.mangle.mangle_response', new=notouch_mangle_rsp)
|
||||
|
||||
def test_proxy_client_mangle_req(mocker, proxy_connection, in_scope_true):
|
||||
# Make the connection
|
||||
(prot, sent, resp_deferred) = yield proxy_connection('GET / HTTP/1.1\r\n\r\n')
|
||||
assert sent == 'GET /mangled HTTP/1.1\r\n\r\n'
|
||||
(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_basic(mocker, proxy_connection):
|
||||
mocker.patch('pappyproxy.mangle.mangle_request', new=notouch_mangle_req)
|
||||
mocker.patch('pappyproxy.mangle.mangle_response', new=rsp_mangler_change)
|
||||
def test_proxy_client_mangle_rsp(mocker, proxy_connection, in_scope_true):
|
||||
# Make the connection
|
||||
(prot, sent, resp_deferred) = yield proxy_connection('GET / HTTP/1.1\r\n\r\n')
|
||||
(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('')
|
||||
ret_req = yield resp_deferred
|
||||
response = ret_req.response.full_response
|
||||
req = yield retreq_deferred
|
||||
response = req.response.full_response
|
||||
assert response == 'HTTP/1.1 500 MANGLED\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'
|
||||
|
|
|
|||
104
pappyproxy/tests/test_session.py
Normal file
104
pappyproxy/tests/test_session.py
Normal file
|
|
@ -0,0 +1,104 @@
|
|||
import pytest
|
||||
from pappyproxy.session import Session
|
||||
from pappyproxy.http import Request, Response, ResponseCookie
|
||||
|
||||
@pytest.fixture
|
||||
def req():
|
||||
r = Request()
|
||||
r.status_line = 'GET / HTTP/1.1'
|
||||
return r
|
||||
|
||||
@pytest.fixture
|
||||
def rsp():
|
||||
r = Response()
|
||||
r.status_line = 'HTTP/1.1 200 OK'
|
||||
return r
|
||||
|
||||
def test_session_basic(req, rsp):
|
||||
s = Session(
|
||||
cookie_vals={'session':'foo'},
|
||||
header_vals={'auth':'bar'},
|
||||
)
|
||||
|
||||
assert 'session' not in req.cookies
|
||||
assert 'session' not in rsp.cookies
|
||||
assert 'auth' not in req.headers
|
||||
assert 'auth' not in rsp.headers
|
||||
s.apply_req(req)
|
||||
s.apply_rsp(rsp)
|
||||
assert req.cookies['session'] == 'foo'
|
||||
assert rsp.cookies['session'].cookie_str == 'session=foo'
|
||||
assert req.headers['auth'] == 'bar'
|
||||
assert 'auth' not in rsp.headers
|
||||
|
||||
def test_session_cookieobj_basic(req, rsp):
|
||||
s = Session(
|
||||
cookie_vals={'session':ResponseCookie('session=foo; secure; httponly; path=/')},
|
||||
header_vals={'auth':'bar'},
|
||||
)
|
||||
|
||||
s.apply_req(req)
|
||||
s.apply_rsp(rsp)
|
||||
assert req.cookies['session'] == 'foo'
|
||||
assert rsp.cookies['session'].key == 'session'
|
||||
assert rsp.cookies['session'].val == 'foo'
|
||||
assert rsp.cookies['session'].secure
|
||||
assert rsp.cookies['session'].http_only
|
||||
assert rsp.cookies['session'].path == '/'
|
||||
assert req.headers['auth'] == 'bar'
|
||||
assert 'auth' not in rsp.headers
|
||||
|
||||
def test_session_get_req(req):
|
||||
req.headers['BasicAuth'] = 'asdfasdf'
|
||||
req.headers['Host'] = 'www.myfavoritecolor.foobar'
|
||||
req.cookies['session'] = 'foobar'
|
||||
req.cookies['favorite_color'] = 'blue'
|
||||
|
||||
s = Session()
|
||||
s.get_req(req, ['session'], ['BasicAuth'])
|
||||
assert s.cookies == ['session']
|
||||
assert s.headers == ['BasicAuth']
|
||||
assert s.cookie_vals['session'].val == 'foobar'
|
||||
assert s.header_vals['BasicAuth'] == 'asdfasdf'
|
||||
assert 'Host' not in s.headers
|
||||
assert 'favorite_color' not in s.cookies
|
||||
|
||||
def test_session_get_rsp(rsp):
|
||||
rsp.headers['BasicAuth'] = 'asdfasdf'
|
||||
rsp.headers['Host'] = 'www.myfavoritecolor.foobar'
|
||||
rsp.set_cookie(ResponseCookie('session=foobar; secure; path=/'))
|
||||
rsp.set_cookie(ResponseCookie('favorite_color=blue; secure; path=/'))
|
||||
|
||||
s = Session()
|
||||
s.get_rsp(rsp, ['session'])
|
||||
assert s.cookies == ['session']
|
||||
assert s.headers == []
|
||||
assert s.cookie_vals['session'].key == 'session'
|
||||
assert s.cookie_vals['session'].val == 'foobar'
|
||||
assert s.cookie_vals['session'].path == '/'
|
||||
assert s.cookie_vals['session'].secure
|
||||
|
||||
def test_session_mixed(req, rsp):
|
||||
s = Session(
|
||||
cookie_names=['session', 'state'],
|
||||
cookie_vals={'session':ResponseCookie('session=foo; secure; httponly; path=/')},
|
||||
header_vals={'auth':'bar'},
|
||||
)
|
||||
|
||||
s.apply_req(req)
|
||||
s.apply_rsp(rsp)
|
||||
assert req.cookies['session'] == 'foo'
|
||||
assert rsp.cookies['session'].key == 'session'
|
||||
assert rsp.cookies['session'].val == 'foo'
|
||||
assert rsp.cookies['session'].secure
|
||||
assert rsp.cookies['session'].http_only
|
||||
assert rsp.cookies['session'].path == '/'
|
||||
assert 'auth' not in rsp.headers
|
||||
|
||||
r = Response()
|
||||
r.status_line = 'HTTP/1.1 200 OK'
|
||||
r.set_cookie(ResponseCookie('state=bazzers'))
|
||||
r.set_cookie(ResponseCookie('session=buzzers'))
|
||||
s.get_rsp(r)
|
||||
assert s.cookie_vals['session'].val == 'buzzers'
|
||||
assert s.cookie_vals['state'].val == 'bazzers'
|
||||
|
|
@ -1,6 +1,11 @@
|
|||
import __builtin__
|
||||
import mock
|
||||
import pytest
|
||||
import StringIO
|
||||
from twisted.internet import defer
|
||||
|
||||
next_mock_id = 0
|
||||
|
||||
class ClassDeleted():
|
||||
pass
|
||||
|
||||
|
|
@ -40,3 +45,31 @@ def no_database(mocker):
|
|||
# Don't make database queries
|
||||
mocker.patch("twisted.enterprise.adbapi.ConnectionPool",
|
||||
new=ClassDeleted)
|
||||
|
||||
def fake_save_request(r):
|
||||
global next_mock_id
|
||||
r.reqid = next_mock_id
|
||||
next_mock_id += 1
|
||||
return mock_deferred(None)
|
||||
|
||||
def fake_save_response(r):
|
||||
global next_mock_id
|
||||
r.rspid = next_mock_id
|
||||
next_mock_id += 1
|
||||
return mock_deferred(None)
|
||||
|
||||
@pytest.fixture
|
||||
def fake_saving(mocker):
|
||||
mocker.patch("pappyproxy.http.Request.async_save", new=fake_save_request)
|
||||
mocker.patch("pappyproxy.http.Response.async_save", new=fake_save_response)
|
||||
|
||||
@pytest.fixture
|
||||
def mock_deep_save(mocker, fake_saving):
|
||||
new_deep_save = mock.MagicMock()
|
||||
new_deep_save.return_value = mock_deferred(None)
|
||||
mocker.patch("pappyproxy.http.Request.async_deep_save", new=new_deep_save)
|
||||
return new_deep_save
|
||||
|
||||
def print_fuck(*args, **kwargs):
|
||||
print 'fuck'
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue