proxy.plugin package

Submodules

• proxy.plugin.cloudflare_dns module
  • proxy.py
• proxy.plugin.custom_dns_resolver module
  • proxy.py
• proxy.plugin.filter_by_client_ip module
  • proxy.py
• proxy.plugin.filter_by_upstream module
  • proxy.py
• proxy.plugin.filter_by_url_regex module
  • proxy.py
• proxy.plugin.grout_client module
  • proxy.py
• proxy.plugin.man_in_the_middle module
  • proxy.py
• proxy.plugin.mock_rest_api module
  • proxy.py
• proxy.plugin.modify_chunk_response module
  • proxy.py
• proxy.plugin.modify_post_data module
  • proxy.py
• proxy.plugin.modify_request_header module
  • proxy.py
• proxy.plugin.program_name module
  • proxy.py
• proxy.plugin.proxy_pool module
  • proxy.py
• proxy.plugin.redirect_to_custom_server module
  • proxy.py
• proxy.plugin.reverse_proxy module
  • proxy.py
• proxy.plugin.shortlink module
  • proxy.py
• proxy.plugin.tls_intercept_conditionally module
  • proxy.py
• proxy.plugin.web_server_route module
  • proxy.py

Module contents

proxy.py

⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on Network monitoring, controls & Application development, testing, debugging.

copyright

3. 2013-present by Abhinav Singh and contributors.

license

BSD, see LICENSE for more details.

class proxy.plugin.BaseCacheResponsesPlugin(*args: Any, **kwargs: Any)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Base cache plugin.

It requires a storage backend to work with. Storage class must implement CacheStore interface.

Different storage backends can be used per request if required.

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called before dispatching client request to upstream.

Note: For pipelined (keep-alive) connections, this handler can be called multiple times, for each request sent to upstream.

Note: If TLS interception is enabled, this handler can be called multiple times if client exchanges multiple requests over same SSL session.

Return optionally modified request object to dispatch to upstream. Return None to drop the request data, e.g. in case a response has already been queued. Raise HttpRequestRejected or HttpProtocolException directly to tear down the connection with client.

handle_upstream_chunk(chunk: memoryview) → Optional[memoryview]

Handler called right after receiving raw response from upstream server.

For HTTPS connections, chunk will be encrypted unless TLS interception is also enabled.

Return None if you don’t want to sent this chunk to the client.

on_upstream_connection_close() → None

Handler called right after upstream connection has been closed.

set_store(store: proxy.plugin.cache.store.base.CacheStore) → None

class proxy.plugin.CacheResponsesPlugin(*args: Any, **kwargs: Any)

Bases: proxy.plugin.cache.base.BaseCacheResponsesPlugin

Caches response using OnDiskCacheStore.

ENABLED = <multiprocessing.synchronize.Event object>

_abc_impl = <_abc._abc_data object>

on_access_log(context: Dict[str, Any]) → Optional[Dict[str, Any]]

Use this method to override default access log format (see DEFAULT_HTTP_ACCESS_LOG_FORMAT and DEFAULT_HTTPS_ACCESS_LOG_FORMAT) and to add/update/modify/delete context for next plugin.on_access_log invocation.

This is specially useful if a plugins want to provide extra context in the access log which may not available within other plugins’ context or even in proxy.py core.

Returns Log context or None. If plugin chooses to access log, they ideally must return None to prevent other plugin.on_access_log invocation.

on_upstream_connection_close() → None

Handler called right after upstream connection has been closed.

static write_content_type(cache_file_path: str, cache_dir: str, content_file_name: str, cache_requests: bool) → Optional[str]

class proxy.plugin.CloudflareDnsResolverPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

This plugin uses Cloudflare DNS resolver to provide protection against malware and adult content. Implementation uses DoH specification.

See https://developers.cloudflare.com/1.1.1.1/1.1.1.1-for-families See https://developers.cloudflare.com/1.1.1.1/encrypted-dns/dns-over-https/make-api-requests/dns-json

Note

For this plugin to work, make sure to bypass proxy for 1.1.1.1

Note

This plugin requires additional dependency because DoH mandates a HTTP2 complaint client. Install httpx dependency as:

pip install "httpx[http2]"

_abc_impl = <_abc._abc_data object>

resolve_dns(host: str, port: int) → Tuple[Optional[str], Optional[Tuple[str, int]]]

Resolve upstream server host to an IP address.

Optionally also override the source address to use for connection with upstream server.

For upstream IP: Return None to use default resolver available to the system. Return IP address as string to use your custom resolver.

For source address: Return None to use default source address Return 2-tuple representing (host, port) to use as source address

class proxy.plugin.CustomDnsResolverPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

This plugin demonstrate how to use your own custom DNS resolver.

_abc_impl = <_abc._abc_data object>

resolve_dns(host: str, port: int) → Tuple[Optional[str], Optional[Tuple[str, int]]]

Here we are using in-built python resolver for demonstration.

Ideally you would like to query your custom DNS server or even use DoH to make real sense out of this plugin.

The second parameter returned is None. Return a 2-tuple to configure underlying interface to use for connection to the upstream server.

class proxy.plugin.FilterByClientIpPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Allow only (whitelist) or Drop only (blacklist) traffic by inspecting incoming client IP address.

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

class proxy.plugin.FilterByURLRegexPlugin(*args: Any, **kwargs: Any)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Drops traffic by inspecting request URL and checking against a list of regular expressions. Example, default filter list below can be used as a starting point for filtering ads.

_abc_impl = <_abc._abc_data object>

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called before dispatching client request to upstream.

Note: For pipelined (keep-alive) connections, this handler can be called multiple times, for each request sent to upstream.

Note: If TLS interception is enabled, this handler can be called multiple times if client exchanges multiple requests over same SSL session.

Return optionally modified request object to dispatch to upstream. Return None to drop the request data, e.g. in case a response has already been queued. Raise HttpRequestRejected or HttpProtocolException directly to tear down the connection with client.

class proxy.plugin.FilterByUpstreamHostPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Drop traffic by inspecting upstream host.

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

class proxy.plugin.ManInTheMiddlePlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Modifies upstream server responses.

_abc_impl = <_abc._abc_data object>

handle_upstream_chunk(_chunk: memoryview) → Optional[memoryview]

Handler called right after receiving raw response from upstream server.

For HTTPS connections, chunk will be encrypted unless TLS interception is also enabled.

Return None if you don’t want to sent this chunk to the client.

class proxy.plugin.ModifyChunkResponsePlugin(*args: Any, **kwargs: Any)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Accumulate & modify chunk responses as received from upstream.

DEFAULT_CHUNKS = [b'modify', b'chunk', b'response', b'plugin']

_abc_impl = <_abc._abc_data object>

handle_upstream_chunk(chunk: memoryview) → Optional[memoryview]

Handler called right after receiving raw response from upstream server.

For HTTPS connections, chunk will be encrypted unless TLS interception is also enabled.

Return None if you don’t want to sent this chunk to the client.

class proxy.plugin.ModifyPostDataPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Modify POST request body before sending to upstream server.

Following curl executions will work:

1. Plain curl -v -x localhost:8899 -X POST http://httpbin.org/post -d ‘key=value’

2. Chunked curl -v -x localhost:8899 -X POST -H ‘Transfer-Encoding: chunked’ http://httpbin.org/post -d ‘key=value’

3. Chunked & Compressed echo ‘key=value’ | gzip | curl -v -x localhost:8899 -X POST –data-binary @- -H ‘Transfer-Encoding: chunked’ -H ‘Content-Encoding: gzip’ http://httpbin.org/post

MODIFIED_BODY = b'{"key": "modified"}'

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called before dispatching client request to upstream.

Note: For pipelined (keep-alive) connections, this handler can be called multiple times, for each request sent to upstream.

Note: If TLS interception is enabled, this handler can be called multiple times if client exchanges multiple requests over same SSL session.

Return optionally modified request object to dispatch to upstream. Return None to drop the request data, e.g. in case a response has already been queued. Raise HttpRequestRejected or HttpProtocolException directly to tear down the connection with client.

class proxy.plugin.ModifyRequestHeaderPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Modify request header before sending to upstream server.

_abc_impl = <_abc._abc_data object>

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

NOTE: This is for HTTPS request headers modification when under TLS interception.

For HTTPS requests, modification of request under TLS interception WILL NOT WORK through before_upstream_connection.

class proxy.plugin.ProgramNamePlugin(*args: Any, **kwargs: Any)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Tries to identify the application connecting to the proxy instance. This is only possible when connection itself originates from the same machine where the proxy instance is running.

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

on_access_log(context: Dict[str, Any]) → Optional[Dict[str, Any]]

Use this method to override default access log format (see DEFAULT_HTTP_ACCESS_LOG_FORMAT and DEFAULT_HTTPS_ACCESS_LOG_FORMAT) and to add/update/modify/delete context for next plugin.on_access_log invocation.

This is specially useful if a plugins want to provide extra context in the access log which may not available within other plugins’ context or even in proxy.py core.

Returns Log context or None. If plugin chooses to access log, they ideally must return None to prevent other plugin.on_access_log invocation.

class proxy.plugin.ProposedRestApiPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Mock responses for your upstream REST API.

Used to test and develop client side applications without need of an actual upstream REST API server.

Returns proposed REST API mock responses to the client without establishing upstream connection.

Note: This plugin won’t work if your client is making HTTPS connection to api.example.com.

API_SERVER = b'api.example.com'

REST_API_SPEC = {b'/v1/users/': {'count': 2, 'next': None, 'previous': None, 'results': [{'email': 'you@example.com', 'groups': [], 'url': 'api.example.com/v1/users/1/', 'username': 'admin'}, {'email': 'someone@example.com', 'groups': [], 'url': 'api.example.com/v1/users/2/', 'username': 'someone'}]}}

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called before dispatching client request to upstream.

Note: For pipelined (keep-alive) connections, this handler can be called multiple times, for each request sent to upstream.

Note: If TLS interception is enabled, this handler can be called multiple times if client exchanges multiple requests over same SSL session.

Return optionally modified request object to dispatch to upstream. Return None to drop the request data, e.g. in case a response has already been queued. Raise HttpRequestRejected or HttpProtocolException directly to tear down the connection with client.

class proxy.plugin.ProxyPoolPlugin(*args: Any, **kwargs: Any)

Bases: proxy.core.base.tcp_upstream.TcpUpstreamConnectionHandler, proxy.http.proxy.plugin.HttpProxyBasePlugin

Proxy pool plugin simply acts as a proxy adapter for proxy.py itself.

Imagine this plugin as setting up proxy settings for proxy.py instance itself. All incoming client requests are proxied to configured upstream proxies.

_abc_impl = <_abc._abc_data object>

_select_proxy() → proxy.http.url.Url

Choose a random proxy from the pool.

TODO: Implement your own logic here e.g. round-robin, least connection etc.

access_log(log_attrs: Dict[str, Any]) → None

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Avoids establishing the default connection to upstream server by returning None.

TODO(abhinavsingh): Ideally connection to upstream proxy endpoints must be bootstrapped within it’s own re-usable and garbage collected pool, to avoid establishing a new upstream proxy connection for each client request.

See UpstreamConnectionPool which is a work in progress for SSL cache handling.

handle_client_data(raw: memoryview) → Optional[memoryview]

Only invoked when before_upstream_connection returns None

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Only invoked once after client original proxy request has been received completely.

handle_upstream_chunk(chunk: memoryview) → Optional[memoryview]

Will never be called since we didn’t establish an upstream connection.

handle_upstream_data(raw: memoryview) → None

on_access_log(context: Dict[str, Any]) → Optional[Dict[str, Any]]

Use this method to override default access log format (see DEFAULT_HTTP_ACCESS_LOG_FORMAT and DEFAULT_HTTPS_ACCESS_LOG_FORMAT) and to add/update/modify/delete context for next plugin.on_access_log invocation.

This is specially useful if a plugins want to provide extra context in the access log which may not available within other plugins’ context or even in proxy.py core.

Returns Log context or None. If plugin chooses to access log, they ideally must return None to prevent other plugin.on_access_log invocation.

on_upstream_connection_close() → None

Called when client connection has been closed.

class proxy.plugin.RedirectToCustomServerPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Modifies client request to redirect all incoming requests to a fixed server address.

UPSTREAM_SERVER = b'http://localhost:8899/'

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

class proxy.plugin.ReverseProxyPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.server.plugin.ReverseProxyBasePlugin

This example plugin is equivalent to following Nginx configuration:

```text
location /get {
    proxy_pass http://httpbin.org/get
}
```

Plugin also demonstrates how to write “Python” equivalent for any “Nginx Lua” based configuration i.e. your plugin code will have full control over what do after one of your route has matched.

_abc_impl = <_abc._abc_data object>

handle_route(request: proxy.http.parser.parser.HttpParser, pattern: proxy.common.types.RePattern) → Union[memoryview, proxy.http.url.Url, TcpServerConnection]

For our example dynamic route, we want to simply convert any incoming request to “/get/1” into “/get?id=1” when serving from upstream.

routes() → List[Union[str, Tuple[str, List[bytes]]]]

List of routes registered by plugin.

There are 2 types of routes:

1. Dynamic routes (str): Should be a regular expression

2. Static routes (tuple): Contain 2 elements, a route regular expression and list of upstream urls to serve when the route matches.

Static routes doesn’t require you to implement the handle_route method. Reverse proxy core will automatically pick one of the configured upstream URL and serve it out-of-box.

Dynamic routes are helpful when you want to dynamically match and serve upstream urls. To handle dynamic routes, you must implement the handle_route method, which must return the url to serve.

class proxy.plugin.ShortLinkPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

Add support for short links in your favorite browsers / applications.

Enable ShortLinkPlugin and speed up your daily browsing experience.

Example:: * f/ for facebook.com * g/ for google.com` * ``t/ for twitter.com * y/ for youtube.com * proxy/ for py internal web servers. Customize map below for your taste and need.

Paths are also preserved. E.g. t/imoracle will resolve to my Twitter profile for username imoracle.

SHORT_LINKS = {b'a': b'amazon.com', b'f': b'facebook.com', b'g': b'google.com', b'i': b'instagram.com', b'l': b'linkedin.com', b'proxy': b'localhost:8899', b't': b'twitter.com', b'w': b'web.whatsapp.com', b'y': b'youtube.com'}

_abc_impl = <_abc._abc_data object>

before_upstream_connection(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called just before Proxy upstream connection is established.

Return optionally modified request object. If None is returned, upstream connection won’t be established.

Raise HttpRequestRejected or HttpProtocolException directly to drop the connection.

handle_client_request(request: proxy.http.parser.parser.HttpParser) → Optional[proxy.http.parser.parser.HttpParser]

Handler called before dispatching client request to upstream.

Note: For pipelined (keep-alive) connections, this handler can be called multiple times, for each request sent to upstream.

Note: If TLS interception is enabled, this handler can be called multiple times if client exchanges multiple requests over same SSL session.

Return optionally modified request object to dispatch to upstream. Return None to drop the request data, e.g. in case a response has already been queued. Raise HttpRequestRejected or HttpProtocolException directly to tear down the connection with client.

class proxy.plugin.TlsInterceptConditionallyPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.proxy.plugin.HttpProxyBasePlugin

TLS intercept conditionally.

_abc_impl = <_abc._abc_data object>

do_intercept(request: proxy.http.parser.parser.HttpParser) → bool

By default returns True (only) when necessary flags for TLS interception are passed.

When TLS interception is enabled, plugins can still disable TLS interception by returning False explicitly. This hook will allow you to run proxy instance with TLS interception flags BUT only conditionally enable interception for certain requests.

class proxy.plugin.WebServerPlugin(uid: str, flags: argparse.Namespace, client: proxy.http.connection.HttpClientConnection, event_queue: proxy.core.event.queue.EventQueue, upstream_conn_pool: Optional[UpstreamConnectionPool] = None)

Bases: proxy.http.server.plugin.HttpWebServerBasePlugin

Demonstrates inbuilt web server routing using plugin.

_abc_impl = <_abc._abc_data object>

handle_request(request: proxy.http.parser.parser.HttpParser) → None

Handle the request and serve response.

on_websocket_message(frame: proxy.http.websocket.frame.WebsocketFrame) → None

Open chrome devtools and try using following commands:

Example:

ws = new WebSocket(“ws://localhost:8899/ws-route-example”) ws.onmessage = function(m) { console.log(m); } ws.send(‘hello’)

routes() → List[Tuple[int, str]]

Return List(protocol, path) that this plugin handles.