certusone
/
proxy-model.py
mirror of https://github.com/certusone/proxy-model.py.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Test refactor + Docker image CI (#154) 

* Move tests into individual modules too

* Ensure one test class per file

* Fix docker image after refactoring

* Add github actions workflow for building docker image

* Fix image name

* Setup python required for extracting proxy version

* Version will also require deps
This commit is contained in:
Abhinav Singh 2019-10-29 20:41:39 -07:00 committed by GitHub
parent 75a818d397
commit 3aa1dc2824
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
48 changed files with 480 additions and 330 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.dockerignore
View File

@ -1,6 +1,11 @@
# Ignore everything
**
# Except proxy.py
!proxy.py
# Except proxy
!proxy
!requirements.txt
!setup.py
!README.md
# Ignore __pycache__ directory
proxy/__pycache__

28
.github/workflows/test-docker.yml vendored Normal file
View File

@ -0,0 +1,28 @@
name: Proxy.py Docker
on: [push]
jobs:
build:
runs-on: ${{ matrix.os }}
name: Python ${{ matrix.python }} on ${{ matrix.os }}
strategy:
matrix:
os: [ubuntu-latest]
python: [3.7]
max-parallel: 1
fail-fast: false
steps:
- uses: actions/checkout@v1
- name: Setup Python
uses: actions/setup-python@v1
with:
python-version: ${{ matrix.python }}-dev
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-testing.txt
- name: Build
run: |
make container

1
.github/workflows/test-library.yml vendored
View File

@ -21,6 +21,7 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-testing.txt
- name: Quality Check
run: |

13
.gitignore vendored
View File

@ -1,11 +1,18 @@
.coverage*
.coverage
.idea
.vscode
.project
.pydevproject
.settings
.mypy_cache
.hypothesis
coverage.xml
proxy.py.iml
*.pyc
ca-*.pem
https-*.pem
node_modules
venv
cover
@ -13,7 +20,3 @@ htmlcov
dist
build
proxy.py.egg-info
proxy.py.iml
*.pyc
ca-*.pem
https-*.pem

16
Dockerfile
View File

@ -1,21 +1,25 @@
FROM python:3.7-alpine as base
FROM base as builder
COPY requirements.txt .
RUN pip install --upgrade pip && pip install --install-option="--prefix=/deps" -r requirements.txt
COPY requirements.txt /app/
COPY setup.py /app/
COPY README.md /app/
COPY proxy/ /app/proxy/
WORKDIR /app
RUN pip install --upgrade pip && \
pip install --install-option="--prefix=/deps" .
FROM base
LABEL com.abhinavsingh.name="abhinavsingh/proxy.py" \
com.abhinavsingh.description="⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file" \
com.abhinavsingh.description="⚡⚡⚡Fast, Lightweight, Programmable, TLS interception capable proxy server for Application debugging, testing and development" \
com.abhinavsingh.url="https://github.com/abhinavsingh/proxy.py" \
com.abhinavsingh.vcs-url="https://github.com/abhinavsingh/proxy.py" \
com.abhinavsingh.docker.cmd="docker run -it --rm -p 8899:8899 abhinavsingh/proxy.py"
COPY --from=builder /deps /usr/local
COPY proxy.py /app/
WORKDIR /app
EXPOSE 8899/tcp
ENTRYPOINT [ "./proxy.py" ]
ENTRYPOINT [ "proxy" ]
CMD [ "--hostname=0.0.0.0", \
"--port=8899" ]

107
Makefile
View File

@ -13,42 +13,11 @@ CA_KEY_FILE_PATH := ca-key.pem
CA_CERT_FILE_PATH := ca-cert.pem
CA_SIGNING_KEY_FILE_PATH := ca-signing-key.pem
.PHONY: all clean test package test-release release coverage lint autopep8
.PHONY: all clean-lib test-lib package test-release release coverage lint autopep8
.PHONY: container run-container release-container https-certificates ca-certificates
.PHONY: profile dashboard clean-dashboard
all: clean test
clean:
find . -name '*.pyc' -exec rm -f {} +
find . -name '*.pyo' -exec rm -f {} +
find . -name '*~' -exec rm -f {} +
rm -f .coverage
rm -rf htmlcov
rm -rf dist
rm -rf build
rm -rf proxy.py.egg-info
rm -rf .pytest_cache
test: lint
python -m unittest tests/*.py
package: clean
python setup.py sdist bdist_wheel
test-release: package
twine upload --verbose --repository-url https://test.pypi.org/legacy/ dist/*
release: package
twine upload dist/*
coverage:
pytest --cov=proxy --cov-report=html tests/
open htmlcov/index.html
lint:
flake8 --ignore=W504 --max-line-length=127 proxy/ tests/ benchmark/ plugin_examples/ dashboard/dashboard.py setup.py
mypy --strict --ignore-missing-imports proxy/ tests/ benchmark/ plugin_examples/ dashboard/dashboard.py setup.py
all: clean-lib test-lib
autopep8:
autopep8 --recursive --in-place --aggressive proxy/*.py
@ -59,22 +28,6 @@ autopep8:
autopep8 --recursive --in-place --aggressive dashboard/*.py
autopep8 --recursive --in-place --aggressive setup.py
container:
docker build -t $(LATEST_TAG) -t $(IMAGE_TAG) .
run-container:
docker run -it -p 8899:8899 --rm $(LATEST_TAG)
release-container:
docker push $(IMAGE_TAG)
docker push $(LATEST_TAG)
https-certificates:
# Generate server key
openssl genrsa -out $(HTTPS_KEY_FILE_PATH) 2048
# Generate server certificate
openssl req -new -x509 -days 3650 -key $(HTTPS_KEY_FILE_PATH) -out $(HTTPS_CERT_FILE_PATH)
ca-certificates:
# Generate CA key
openssl genrsa -out $(CA_KEY_FILE_PATH) 2048
@ -84,11 +37,59 @@ ca-certificates:
# Generated certificates are then signed with CA certificate / key generated above
openssl genrsa -out $(CA_SIGNING_KEY_FILE_PATH) 2048
profile:
sudo py-spy -F -f profile.svg -d 3600 proxy.py
clean-lib:
find . -name '*.pyc' -exec rm -f {} +
find . -name '*.pyo' -exec rm -f {} +
find . -name '*~' -exec rm -f {} +
rm -f .coverage
rm -rf htmlcov
rm -rf dist
rm -rf build
rm -rf proxy.py.egg-info
rm -rf .pytest_cache
rm -rf .hypothesis
clean-dashboard:
rm -rf public/dashboard
container:
docker build -t $(LATEST_TAG) -t $(IMAGE_TAG) .
coverage:
pytest --cov=proxy --cov-report=html tests/
open htmlcov/index.html
dashboard:
pushd dashboard && npm run build && popd
clean-dashboard:
rm -rf public/dashboard
https-certificates:
# Generate server key
openssl genrsa -out $(HTTPS_KEY_FILE_PATH) 2048
# Generate server certificate
openssl req -new -x509 -days 3650 -key $(HTTPS_KEY_FILE_PATH) -out $(HTTPS_CERT_FILE_PATH)
lint:
flake8 --ignore=W504 --max-line-length=127 proxy/ tests/ benchmark/ plugin_examples/ dashboard/dashboard.py setup.py
mypy --strict --ignore-missing-imports proxy/ tests/ benchmark/ plugin_examples/ dashboard/dashboard.py setup.py
package: clean
python setup.py sdist bdist_wheel
profile:
sudo py-spy -F -f profile.svg -d 3600 proxy.py
release: package
twine upload dist/*
release-container:
docker push $(IMAGE_TAG)
docker push $(LATEST_TAG)
run-container:
docker run -it -p 8899:8899 --rm $(LATEST_TAG)
test-lib: lint
python -m unittest discover
test-release: package
twine upload --verbose --repository-url https://test.pypi.org/legacy/ dist/*

5
README.md
View File

@ -1119,14 +1119,15 @@ Changelog
- `v2.x`
- No longer ~~a single file module~~.
- Added support for threadless execution.
- Added dashboard app.
- `v1.x`
- `Python3` only.
- Deprecated support for ~~Python 2.x~~.
- Added support for multi accept.
- Added support multi core accept.
- Added plugin support.
- `v0.x`
- Single file.
- Single threaded server.
For detailed changelog refer
For detailed changelog refer either to release PRs or commit history.

3
benchmark/benchmark.py
View File

@ -3,7 +3,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
dashboard/dashboard.py
View File

@ -1,7 +1,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
dashboard/src/devtools.ts
View File

@ -1,7 +1,8 @@
/*
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

9
dashboard/src/proxy.css
View File

@ -1,3 +1,12 @@
/*
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
*/
#app {
background-color: #eeeeee;
height: 100%;

9
dashboard/src/proxy.html
View File

@ -1,3 +1,12 @@
<!--
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
-->
<html lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html;charset=utf-8">

3
dashboard/src/proxy.ts
View File

@ -1,7 +1,8 @@
/*
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

10
dashboard/test/test.ts
View File

@ -1,3 +1,13 @@
/*
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
*/
import { ProxyDashboard } from "../src/proxy";
describe("test suite", () => {

3
helper/Procfile
View File

@ -1,6 +1,7 @@
# proxy.py
# ~~~~~~~~
# ⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
# ⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
# proxy server for Application debugging, testing and development.
#
# :copyright: (c) 2013-present by Abhinav Singh and contributors.
# :license: BSD, see LICENSE for more details.

3
helper/chrome_with_proxy.sh
View File

@ -2,7 +2,8 @@
# proxy.py
# ~~~~~~~~
# ⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
# ⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
# proxy server for Application debugging, testing and development.
#
# :copyright: (c) 2013-present by Abhinav Singh and contributors.
# :license: BSD, see LICENSE for more details.

3
helper/fluentd.conf
View File

@ -1,6 +1,7 @@
# proxy.py
# ~~~~~~~~
# ⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
# ⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
# proxy server for Application debugging, testing and development.
#
# :copyright: (c) 2013-present by Abhinav Singh and contributors.
# :license: BSD, see LICENSE for more details.

3
helper/monitor_open_files.sh
View File

@ -2,7 +2,8 @@
# proxy.py
# ~~~~~~~~
# ⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
# ⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
# proxy server for Application debugging, testing and development.
#
# :copyright: (c) 2013-present by Abhinav Singh and contributors.
# :license: BSD, see LICENSE for more details.

3
plugin_examples/__init__.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/cache_responses.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/filter_by_upstream.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/man_in_the_middle.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/mock_rest_api.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/modify_post_data.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/redirect_to_custom_server.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/shortlink.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
plugin_examples/web_server_route.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

3
tests/__init__.py
View File

@ -2,7 +2,8 @@
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
⚡⚡⚡ Fast, Lightweight, Programmable, TLS interception capable
proxy server for Application debugging, testing and development.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.

9
tests/common/__init__.py Normal file
View File

@ -0,0 +1,9 @@
# -*- coding: utf-8 -*-
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
"""

0
tests/test_text_bytes.py → tests/common/test_text_bytes.py
View File

0
tests/test_utils.py → tests/common/test_utils.py
View File

9
tests/core/__init__.py Normal file
View File

@ -0,0 +1,9 @@
# -*- coding: utf-8 -*-
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
"""

52
tests/test_acceptor.py → tests/core/test_acceptor.py
View File

@ -14,7 +14,7 @@ import multiprocessing
from unittest import mock
from proxy.common.flags import Flags
from proxy.core.acceptor import Acceptor, AcceptorPool
from proxy.core.acceptor import Acceptor
class TestAcceptor(unittest.TestCase):
@ -95,53 +95,3 @@ class TestAcceptor(unittest.TestCase):
target=self.mock_protocol_handler.return_value.run)
mock_thread.return_value.start.assert_called()
sock.close.assert_called()
class TestAcceptorPool(unittest.TestCase):
@mock.patch('proxy.core.acceptor.send_handle')
@mock.patch('multiprocessing.Pipe')
@mock.patch('socket.socket')
@mock.patch('proxy.core.acceptor.Acceptor')
def test_setup_and_shutdown(
self,
mock_worker: mock.Mock,
mock_socket: mock.Mock,
mock_pipe: mock.Mock,
_mock_send_handle: mock.Mock) -> None:
mock_worker1 = mock.MagicMock()
mock_worker2 = mock.MagicMock()
mock_worker.side_effect = [mock_worker1, mock_worker2]
num_workers = 2
sock = mock_socket.return_value
work_klass = mock.MagicMock()
flags = Flags(num_workers=2)
acceptor = AcceptorPool(flags=flags, work_klass=work_klass)
acceptor.setup()
work_klass.assert_not_called()
mock_socket.assert_called_with(
socket.AF_INET6 if acceptor.flags.hostname.version == 6 else socket.AF_INET,
socket.SOCK_STREAM
)
sock.setsockopt.assert_called_with(
socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind.assert_called_with(
(str(acceptor.flags.hostname), acceptor.flags.port))
sock.listen.assert_called_with(acceptor.flags.backlog)
sock.setblocking.assert_called_with(False)
self.assertTrue(mock_pipe.call_count, num_workers)
self.assertTrue(mock_worker.call_count, num_workers)
mock_worker1.start.assert_called()
mock_worker1.join.assert_not_called()
mock_worker2.start.assert_called()
mock_worker2.join.assert_not_called()
sock.close.assert_called()
acceptor.shutdown()
mock_worker1.join.assert_called()
mock_worker2.join.assert_called()

56
tests/core/test_acceptor_pool.py Normal file
View File

@ -0,0 +1,56 @@
import unittest
import socket
from unittest import mock
from proxy.common.flags import Flags
from proxy.core.acceptor import AcceptorPool
class TestAcceptorPool(unittest.TestCase):
@mock.patch('proxy.core.acceptor.send_handle')
@mock.patch('multiprocessing.Pipe')
@mock.patch('socket.socket')
@mock.patch('proxy.core.acceptor.Acceptor')
def test_setup_and_shutdown(
self,
mock_worker: mock.Mock,
mock_socket: mock.Mock,
mock_pipe: mock.Mock,
_mock_send_handle: mock.Mock) -> None:
mock_worker1 = mock.MagicMock()
mock_worker2 = mock.MagicMock()
mock_worker.side_effect = [mock_worker1, mock_worker2]
num_workers = 2
sock = mock_socket.return_value
work_klass = mock.MagicMock()
flags = Flags(num_workers=2)
acceptor = AcceptorPool(flags=flags, work_klass=work_klass)
acceptor.setup()
work_klass.assert_not_called()
mock_socket.assert_called_with(
socket.AF_INET6 if acceptor.flags.hostname.version == 6 else socket.AF_INET,
socket.SOCK_STREAM
)
sock.setsockopt.assert_called_with(
socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind.assert_called_with(
(str(acceptor.flags.hostname), acceptor.flags.port))
sock.listen.assert_called_with(acceptor.flags.backlog)
sock.setblocking.assert_called_with(False)
self.assertTrue(mock_pipe.call_count, num_workers)
self.assertTrue(mock_worker.call_count, num_workers)
mock_worker1.start.assert_called()
mock_worker1.join.assert_not_called()
mock_worker2.start.assert_called()
mock_worker2.join.assert_not_called()
sock.close.assert_called()
acceptor.shutdown()
mock_worker1.join.assert_called()
mock_worker2.join.assert_called()

0
tests/test_connection.py → tests/core/test_connection.py
View File

9
tests/http/__init__.py Normal file
View File

@ -0,0 +1,9 @@
# -*- coding: utf-8 -*-
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
"""

0
tests/test_chunk_parser.py → tests/http/test_chunk_parser.py
View File

0
tests/test_http_parser.py → tests/http/test_http_parser.py
View File

0
tests/test_http_proxy.py → tests/http/test_http_proxy.py
View File

196
tests/test_http_proxy_examples.py → tests/http/test_http_proxy_examples.py
View File

@ -9,45 +9,23 @@
"""
import unittest
import selectors
import ssl
import socket
import json
from urllib import parse as urlparse
from unittest import mock
from typing import Type, cast, Any
from typing import cast
from proxy.common.flags import Flags
from proxy.http.handler import HttpProtocolHandler
from proxy.http.proxy import HttpProxyBasePlugin, HttpProxyPlugin
from proxy.http.proxy import HttpProxyPlugin
from proxy.common.utils import build_http_request, bytes_, build_http_response
from proxy.common.constants import PROXY_AGENT_HEADER_VALUE
from proxy.http.codes import httpStatusCodes
from proxy.http.methods import httpMethods
from plugin_examples import modify_post_data
from plugin_examples import mock_rest_api
from plugin_examples import redirect_to_custom_server
from plugin_examples import filter_by_upstream
from plugin_examples import cache_responses
from plugin_examples import man_in_the_middle
def get_plugin_by_test_name(test_name: str) -> Type[HttpProxyBasePlugin]:
plugin: Type[HttpProxyBasePlugin] = modify_post_data.ModifyPostDataPlugin
if test_name == 'test_modify_post_data_plugin':
plugin = modify_post_data.ModifyPostDataPlugin
elif test_name == 'test_proposed_rest_api_plugin':
plugin = mock_rest_api.ProposedRestApiPlugin
elif test_name == 'test_redirect_to_custom_server_plugin':
plugin = redirect_to_custom_server.RedirectToCustomServerPlugin
elif test_name == 'test_filter_by_upstream_host_plugin':
plugin = filter_by_upstream.FilterByUpstreamHostPlugin
elif test_name == 'test_cache_responses_plugin':
plugin = cache_responses.CacheResponsesPlugin
elif test_name == 'test_man_in_the_middle_plugin':
plugin = man_in_the_middle.ManInTheMiddlePlugin
return plugin
from .utils import get_plugin_by_test_name
class TestHttpProxyPluginExamples(unittest.TestCase):
@ -274,171 +252,3 @@ class TestHttpProxyPluginExamples(unittest.TestCase):
httpStatusCodes.OK,
reason=b'OK', body=b'Hello from man in the middle')
)
class TestHttpProxyPluginExamplesWithTlsInterception(unittest.TestCase):
@mock.patch('ssl.wrap_socket')
@mock.patch('ssl.create_default_context')
@mock.patch('proxy.http.proxy.TcpServerConnection')
@mock.patch('subprocess.Popen')
@mock.patch('selectors.DefaultSelector')
@mock.patch('socket.fromfd')
def setUp(self,
mock_fromfd: mock.Mock,
mock_selector: mock.Mock,
mock_popen: mock.Mock,
mock_server_conn: mock.Mock,
mock_ssl_context: mock.Mock,
mock_ssl_wrap: mock.Mock) -> None:
self.mock_fromfd = mock_fromfd
self.mock_selector = mock_selector
self.mock_popen = mock_popen
self.mock_server_conn = mock_server_conn
self.mock_ssl_context = mock_ssl_context
self.mock_ssl_wrap = mock_ssl_wrap
self.fileno = 10
self._addr = ('127.0.0.1', 54382)
self.flags = Flags(
ca_cert_file='ca-cert.pem',
ca_key_file='ca-key.pem',
ca_signing_key_file='ca-signing-key.pem',)
self.plugin = mock.MagicMock()
plugin = get_plugin_by_test_name(self._testMethodName)
self.flags.plugins = {
b'HttpProtocolHandlerPlugin': [HttpProxyPlugin],
b'HttpProxyBasePlugin': [plugin],
}
self._conn = mock.MagicMock(spec=socket.socket)
mock_fromfd.return_value = self._conn
self.protocol_handler = HttpProtocolHandler(
self.fileno, self._addr, flags=self.flags)
self.protocol_handler.initialize()
self.server = self.mock_server_conn.return_value
self.server_ssl_connection = mock.MagicMock(spec=ssl.SSLSocket)
self.mock_ssl_context.return_value.wrap_socket.return_value = self.server_ssl_connection
self.client_ssl_connection = mock.MagicMock(spec=ssl.SSLSocket)
self.mock_ssl_wrap.return_value = self.client_ssl_connection
def has_buffer() -> bool:
return cast(bool, self.server.queue.called)
def closed() -> bool:
return not self.server.connect.called
def mock_connection() -> Any:
if self.mock_ssl_context.return_value.wrap_socket.called:
return self.server_ssl_connection
return self._conn
self.server.has_buffer.side_effect = has_buffer
type(self.server).closed = mock.PropertyMock(side_effect=closed)
type(
self.server).connection = mock.PropertyMock(
side_effect=mock_connection)
self.mock_selector.return_value.select.side_effect = [
[(selectors.SelectorKey(
fileobj=self._conn,
fd=self._conn.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)],
[(selectors.SelectorKey(
fileobj=self.client_ssl_connection,
fd=self.client_ssl_connection.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)],
[(selectors.SelectorKey(
fileobj=self.server_ssl_connection,
fd=self.server_ssl_connection.fileno,
events=selectors.EVENT_WRITE,
data=None), selectors.EVENT_WRITE)],
[(selectors.SelectorKey(
fileobj=self.server_ssl_connection,
fd=self.server_ssl_connection.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)], ]
# Connect
def send(raw: bytes) -> int:
return len(raw)
self._conn.send.side_effect = send
self._conn.recv.return_value = build_http_request(
httpMethods.CONNECT, b'uni.corn:443'
)
self.protocol_handler.run_once()
self.mock_popen.assert_called()
self.mock_server_conn.assert_called_once_with('uni.corn', 443)
self.server.connect.assert_called()
self.assertEqual(
self.protocol_handler.client.connection,
self.client_ssl_connection)
self.assertEqual(self.server.connection, self.server_ssl_connection)
self._conn.send.assert_called_with(
HttpProxyPlugin.PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT
)
self.assertEqual(self.protocol_handler.client.buffer, b'')
def test_modify_post_data_plugin(self) -> None:
original = b'{"key": "value"}'
modified = b'{"key": "modified"}'
self.client_ssl_connection.recv.return_value = build_http_request(
b'POST', b'/',
headers={
b'Host': b'uni.corn',
b'Content-Type': b'application/x-www-form-urlencoded',
b'Content-Length': bytes_(len(original)),
},
body=original
)
self.protocol_handler.run_once()
self.server.queue.assert_called_with(
build_http_request(
b'POST', b'/',
headers={
b'Host': b'uni.corn',
b'Content-Length': bytes_(len(modified)),
b'Content-Type': b'application/json',
},
body=modified
)
)
@mock.patch('proxy.http.proxy.TcpServerConnection')
def test_man_in_the_middle_plugin(
self, mock_server_conn: mock.Mock) -> None:
request = build_http_request(
b'GET', b'/',
headers={
b'Host': b'uni.corn',
}
)
self.client_ssl_connection.recv.return_value = request
# Client read
self.protocol_handler.run_once()
self.server.queue.assert_called_once_with(request)
# Server write
self.protocol_handler.run_once()
self.server.flush.assert_called_once()
# Server read
self.server.recv.return_value = \
build_http_response(
httpStatusCodes.OK,
reason=b'OK', body=b'Original Response From Upstream')
self.protocol_handler.run_once()
self.assertEqual(
self.protocol_handler.client.buffer,
build_http_response(
httpStatusCodes.OK,
reason=b'OK', body=b'Hello from man in the middle')
)

192
tests/http/test_http_proxy_examples_with_tls_interception.py Normal file
View File

@ -0,0 +1,192 @@
# -*- coding: utf-8 -*-
"""
proxy.py
~~~~~~~~
⚡⚡⚡ Fast, Lightweight, Programmable Proxy Server in a single Python file.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
"""
import unittest
import socket
import selectors
import ssl
from unittest import mock
from typing import Any, cast
from proxy.common.utils import bytes_
from proxy.common.flags import Flags
from proxy.common.utils import build_http_request, build_http_response
from proxy.http.codes import httpStatusCodes
from proxy.http.methods import httpMethods
from proxy.http.handler import HttpProtocolHandler
from proxy.http.proxy import HttpProxyPlugin
from .utils import get_plugin_by_test_name
class TestHttpProxyPluginExamplesWithTlsInterception(unittest.TestCase):
@mock.patch('ssl.wrap_socket')
@mock.patch('ssl.create_default_context')
@mock.patch('proxy.http.proxy.TcpServerConnection')
@mock.patch('subprocess.Popen')
@mock.patch('selectors.DefaultSelector')
@mock.patch('socket.fromfd')
def setUp(self,
mock_fromfd: mock.Mock,
mock_selector: mock.Mock,
mock_popen: mock.Mock,
mock_server_conn: mock.Mock,
mock_ssl_context: mock.Mock,
mock_ssl_wrap: mock.Mock) -> None:
self.mock_fromfd = mock_fromfd
self.mock_selector = mock_selector
self.mock_popen = mock_popen
self.mock_server_conn = mock_server_conn
self.mock_ssl_context = mock_ssl_context
self.mock_ssl_wrap = mock_ssl_wrap
self.fileno = 10
self._addr = ('127.0.0.1', 54382)
self.flags = Flags(
ca_cert_file='ca-cert.pem',
ca_key_file='ca-key.pem',
ca_signing_key_file='ca-signing-key.pem',)
self.plugin = mock.MagicMock()
plugin = get_plugin_by_test_name(self._testMethodName)
self.flags.plugins = {
b'HttpProtocolHandlerPlugin': [HttpProxyPlugin],
b'HttpProxyBasePlugin': [plugin],
}
self._conn = mock.MagicMock(spec=socket.socket)
mock_fromfd.return_value = self._conn
self.protocol_handler = HttpProtocolHandler(
self.fileno, self._addr, flags=self.flags)
self.protocol_handler.initialize()
self.server = self.mock_server_conn.return_value
self.server_ssl_connection = mock.MagicMock(spec=ssl.SSLSocket)
self.mock_ssl_context.return_value.wrap_socket.return_value = self.server_ssl_connection
self.client_ssl_connection = mock.MagicMock(spec=ssl.SSLSocket)
self.mock_ssl_wrap.return_value = self.client_ssl_connection
def has_buffer() -> bool:
return cast(bool, self.server.queue.called)
def closed() -> bool:
return not self.server.connect.called
def mock_connection() -> Any:
if self.mock_ssl_context.return_value.wrap_socket.called:
return self.server_ssl_connection
return self._conn
self.server.has_buffer.side_effect = has_buffer
type(self.server).closed = mock.PropertyMock(side_effect=closed)
type(
self.server).connection = mock.PropertyMock(
side_effect=mock_connection)
self.mock_selector.return_value.select.side_effect = [
[(selectors.SelectorKey(
fileobj=self._conn,
fd=self._conn.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)],
[(selectors.SelectorKey(
fileobj=self.client_ssl_connection,
fd=self.client_ssl_connection.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)],
[(selectors.SelectorKey(
fileobj=self.server_ssl_connection,
fd=self.server_ssl_connection.fileno,
events=selectors.EVENT_WRITE,
data=None), selectors.EVENT_WRITE)],
[(selectors.SelectorKey(
fileobj=self.server_ssl_connection,
fd=self.server_ssl_connection.fileno,
events=selectors.EVENT_READ,
data=None), selectors.EVENT_READ)], ]
# Connect
def send(raw: bytes) -> int:
return len(raw)
self._conn.send.side_effect = send
self._conn.recv.return_value = build_http_request(
httpMethods.CONNECT, b'uni.corn:443'
)
self.protocol_handler.run_once()
self.mock_popen.assert_called()
self.mock_server_conn.assert_called_once_with('uni.corn', 443)
self.server.connect.assert_called()
self.assertEqual(
self.protocol_handler.client.connection,
self.client_ssl_connection)
self.assertEqual(self.server.connection, self.server_ssl_connection)
self._conn.send.assert_called_with(
HttpProxyPlugin.PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT
)
self.assertEqual(self.protocol_handler.client.buffer, b'')
def test_modify_post_data_plugin(self) -> None:
original = b'{"key": "value"}'
modified = b'{"key": "modified"}'
self.client_ssl_connection.recv.return_value = build_http_request(
b'POST', b'/',
headers={
b'Host': b'uni.corn',
b'Content-Type': b'application/x-www-form-urlencoded',
b'Content-Length': bytes_(len(original)),
},
body=original
)
self.protocol_handler.run_once()
self.server.queue.assert_called_with(
build_http_request(
b'POST', b'/',
headers={
b'Host': b'uni.corn',
b'Content-Length': bytes_(len(modified)),
b'Content-Type': b'application/json',
},
body=modified
)
)
def test_man_in_the_middle_plugin(self) -> None:
request = build_http_request(
b'GET', b'/',
headers={
b'Host': b'uni.corn',
}
)
self.client_ssl_connection.recv.return_value = request
# Client read
self.protocol_handler.run_once()
self.server.queue.assert_called_once_with(request)
# Server write
self.protocol_handler.run_once()
self.server.flush.assert_called_once()
# Server read
self.server.recv.return_value = \
build_http_response(
httpStatusCodes.OK,
reason=b'OK', body=b'Original Response From Upstream')
self.protocol_handler.run_once()
self.assertEqual(
self.protocol_handler.client.buffer,
build_http_response(
httpStatusCodes.OK,
reason=b'OK', body=b'Hello from man in the middle')
)

0
tests/test_http_proxy_tls_interception.py → tests/http/test_http_proxy_tls_interception.py
View File

0
tests/test_http_request_rejected.py → tests/http/test_http_request_rejected.py
View File

0
tests/test_protocol_handler.py → tests/http/test_protocol_handler.py
View File

0
tests/test_web_server.py → tests/http/test_web_server.py
View File

0
tests/test_websocket_client.py → tests/http/test_websocket_client.py
View File

0
tests/test_websocket_frame.py → tests/http/test_websocket_frame.py
View File

26
tests/http/utils.py Normal file
View File

@ -0,0 +1,26 @@
from typing import Type
from proxy.http.proxy import HttpProxyBasePlugin
from plugin_examples import modify_post_data
from plugin_examples import mock_rest_api
from plugin_examples import redirect_to_custom_server
from plugin_examples import filter_by_upstream
from plugin_examples import cache_responses
from plugin_examples import man_in_the_middle
def get_plugin_by_test_name(test_name: str) -> Type[HttpProxyBasePlugin]:
plugin: Type[HttpProxyBasePlugin] = modify_post_data.ModifyPostDataPlugin
if test_name == 'test_modify_post_data_plugin':
plugin = modify_post_data.ModifyPostDataPlugin
elif test_name == 'test_proposed_rest_api_plugin':
plugin = mock_rest_api.ProposedRestApiPlugin
elif test_name == 'test_redirect_to_custom_server_plugin':
plugin = redirect_to_custom_server.RedirectToCustomServerPlugin
elif test_name == 'test_filter_by_upstream_host_plugin':
plugin = filter_by_upstream.FilterByUpstreamHostPlugin
elif test_name == 'test_cache_responses_plugin':
plugin = cache_responses.CacheResponsesPlugin
elif test_name == 'test_man_in_the_middle_plugin':
plugin = man_in_the_middle.ManInTheMiddlePlugin
return plugin