Rob Glew
8 years ago
parent
9648bc44cc
commit
d2f0e5c222
17 changed files with 640 additions and 1076 deletions
@ -1 +1 @@ |
||||
__version__ = '0.2.9' |
||||
__version__ = '0.2.10' |
||||
|
||||
@ -1,209 +0,0 @@ |
||||
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') |
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue