v2.4.0 (#1081)

* Update description for `--max-sendbuf-size`

* Add `--cache-by-content-type` flag (#1070)

Add `--cache-by-content-type` flag

* Enhance content type cache (#1072)

* Fix dispatcher exception during unsubs (#1073)

* [EventDispatcher] Guard against broken pipe and eof (#1074)

* Guard against broken pipe and eof

* refactor

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

* [EventCore] Use `no_wait` when publishing to queues (#1076)

* Use `no_wait` when publishing to the event core queues

* Update readme flag version

* [Examples] Fix broken examples (#1077)

* Fix broken examples

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix mypy

* Pylint ignore

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

* Allow multiport with ephemeral & unix socket support (#1078)

* Allow multiport with ephemeral & unix socket support

* Fix unix tests

* `Run in detached (background) mode` instructions

* Update README.md

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: abhinavsingh

Makefile

@@ -7,15 +7,17 @@ PROXYPY_CONTAINER_VERSION := latest
 # Used by container build and run targets
 PROXYPY_CONTAINER_TAG := $(NS)/$(IMAGE_NAME):$(PROXYPY_CONTAINER_VERSION)
 
-HTTPS_KEY_FILE_PATH := https-key.pem
-HTTPS_CERT_FILE_PATH := https-cert.pem
-HTTPS_CSR_FILE_PATH := https-csr.pem
-HTTPS_SIGNED_CERT_FILE_PATH := https-signed-cert.pem
+CERT_DIR :=
+
+HTTPS_KEY_FILE_PATH := $(CERT_DIR)https-key.pem
+HTTPS_CERT_FILE_PATH := $(CERT_DIR)https-cert.pem
+HTTPS_CSR_FILE_PATH := $(CERT_DIR)https-csr.pem
+HTTPS_SIGNED_CERT_FILE_PATH := $(CERT_DIR)https-signed-cert.pem
 
 CA_CERT_SUFFIX :=
-CA_KEY_FILE_PATH := ca-key$(CA_CERT_SUFFIX).pem
-CA_CERT_FILE_PATH := ca-cert$(CA_CERT_SUFFIX).pem
-CA_SIGNING_KEY_FILE_PATH := ca-signing-key$(CA_CERT_SUFFIX).pem
+CA_KEY_FILE_PATH := $(CERT_DIR)ca-key$(CA_CERT_SUFFIX).pem
+CA_CERT_FILE_PATH := $(CERT_DIR)ca-cert$(CA_CERT_SUFFIX).pem
+CA_SIGNING_KEY_FILE_PATH := $(CERT_DIR)ca-signing-key$(CA_CERT_SUFFIX).pem
 
 # Dummy invalid hardcoded value
 PROXYPY_PKG_PATH := dist/proxy.py.whl

README.md

@@ -2286,8 +2286,9 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
           [--ca-key-file CA_KEY_FILE] [--ca-cert-dir CA_CERT_DIR]
           [--ca-cert-file CA_CERT_FILE] [--ca-file CA_FILE]
           [--ca-signing-key-file CA_SIGNING_KEY_FILE]
-          [--auth-plugin AUTH_PLUGIN] [--cache-dir CACHE_DIR]
-          [--cache-requests] [--proxy-pool PROXY_POOL] [--enable-web-server]
+          [--auth-plugin AUTH_PLUGIN] [--cache-requests]
+          [--cache-by-content-type] [--cache-dir CACHE_DIR]
+          [--proxy-pool PROXY_POOL] [--enable-web-server]
           [--enable-static-server] [--static-server-dir STATIC_SERVER_DIR]
           [--min-compression-length MIN_COMPRESSION_LENGTH]
           [--enable-reverse-proxy] [--pac-file PAC_FILE]
@@ -2297,7 +2298,7 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
           [--filtered-client-ips FILTERED_CLIENT_IPS]
           [--filtered-url-regex-config FILTERED_URL_REGEX_CONFIG]
 
-proxy.py v2.4.0rc9.dev8+gea0253d.d20220126
+proxy.py v2.4.0rc10.dev13+g96428ae.d20220126
 
 options:
   -h, --help            show this help message and exit
@@ -2393,8 +2394,8 @@ options:
                         Default: 128 KB. Maximum amount of data received from
                         the server in a single recv() operation.
   --max-sendbuf-size MAX_SENDBUF_SIZE
-                        Default: 64 KB. Maximum amount of data to dispatch in
-                        a single send() operation.
+                        Default: 64 KB. Maximum amount of data to flush in a
+                        single send() operation.
   --timeout TIMEOUT     Default: 10.0. Number of seconds after which an
                         inactive connection must be dropped. Inactivity is
                         defined by no data sent or received by the client.
@@ -2427,11 +2428,16 @@ options:
   --auth-plugin AUTH_PLUGIN
                         Default: proxy.http.proxy.auth.AuthPlugin. Auth plugin
                         to use instead of default basic auth plugin.
+  --cache-requests      Default: False. Whether to also write request packets
+                        in the cache file.
+  --cache-by-content-type
+                        Default: False. Whether to extract content by type
+                        from responses. Extracted content type is written to
+                        the cache directory e.g. video.mp4.
   --cache-dir CACHE_DIR
                         Default: /Users/abhinavsingh/.proxy/cache. Flag only
                         applicable when cache plugin is used with on-disk
                         storage.
-  --cache-requests      Default: False. Whether to also cache request packets.
   --proxy-pool PROXY_POOL
                         List of upstream proxies to use in the pool
   --enable-web-server   Default: False. Whether to enable

examples/pubsub_eventing.py

@@ -66,42 +66,37 @@ def publisher_process(
         # Create a subscriber.
         # Internally, subscribe will start a separate thread
         # to receive incoming published messages.
-        subscriber = EventSubscriber(event_manager.queue, callback=on_event)
-        subscriber.setup()
-
-        # Start a publisher process to demonstrate safe exchange
-        # of messages between processes.
-        publisher_shutdown_event = multiprocessing.Event()
-        publisher = multiprocessing.Process(
-            target=publisher_process, args=(
-                publisher_shutdown_event, event_manager.queue, ),
-        )
-        publisher.start()
-
-        # Dispatch event from main process too
-        # to demonstrate safe exchange of messages
-        # between threads.
-        try:
-            while True:
-                event_manager.queue.publish(
-                    request_id='1234',
-                    event_name=eventNames.WORK_STARTED,
-                    event_payload={'time': time.time()},
-                    publisher_id='eventing_pubsub_main',
-                )
-        except KeyboardInterrupt:
-            logger.info('bye!!!')
-        finally:
-            # Stop publisher process
-            publisher_shutdown_event.set()
-            publisher.join()
-            # Stop subscriber thread
-            subscriber.unsubscribe()
-            logger.info(
-                'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
-                    num_events_received[0], num_events_received[1], time.time(
-                    ) - start_time,
-                ),
+        with EventSubscriber(event_manager.queue, callback=on_event) as subscriber:
+            # Start a publisher process to demonstrate safe exchange
+            # of messages between processes.
+            publisher_shutdown_event = multiprocessing.Event()
+            publisher = multiprocessing.Process(
+                target=publisher_process, args=(
+                    publisher_shutdown_event, event_manager.queue, ),
             )
-    if subscriber:
-        subscriber.shutdown(do_unsubscribe=False)
+            publisher.start()
+
+            # Dispatch event from main process too
+            # to demonstrate safe exchange of messages
+            # between threads.
+            try:
+                while True:
+                    event_manager.queue.publish(
+                        request_id='1234',
+                        event_name=eventNames.WORK_STARTED,
+                        event_payload={'time': time.time()},
+                        publisher_id='eventing_pubsub_main',
+                    )
+            except KeyboardInterrupt:
+                logger.info('KBE!!!')
+            finally:
+                # Stop publisher process
+                publisher_shutdown_event.set()
+                publisher.join()
+                logger.info(
+                    'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
+                        num_events_received[0], num_events_received[1], time.time(
+                        ) - start_time,
+                    ),
+                )
+    logger.info('Done!!!')

examples/ssl_echo_client.py

@@ -8,21 +8,29 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
+import ssl
 import logging
 
 from proxy.core.connection import TcpServerConnection
-from proxy.common.constants import DEFAULT_BUFFER_SIZE
+from proxy.common.constants import DEFAULT_LOG_FORMAT, DEFAULT_BUFFER_SIZE
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 logger = logging.getLogger(__name__)
 
 if __name__ == '__main__':
-    client = TcpServerConnection('::', 12345)
+    client = TcpServerConnection('127.0.0.1', 12345)
     client.connect()
-    client.wrap('example.com', ca_file='ca-cert.pem')
-    # wrap() will by default set connection to nonblocking
-    # flip it back to blocking
-    client.connection.setblocking(True)
+    client.wrap(
+        None,  # 'localhost',
+        ca_file='ca-cert.pem',
+        # For self-signed certs you will have
+        # to disable verification.  Or you can
+        # add your CA certificate in the CA bundle
+        # and then enable verify.
+        verify_mode=ssl.VerifyMode.CERT_NONE,
+    )
     try:
         while True:
             client.send(b'hello')

examples/ssl_echo_server.py

@@ -52,8 +52,8 @@ def main() -> None:
         threadless=True,
         num_workers=1,
         port=12345,
-        keyfile='https-key.pem',
-        certfile='https-signed-cert.pem',
+        key_file='https-key.pem',
+        cert_file='https-signed-cert.pem',
     ):
         try:
             while True:

examples/tcp_echo_client.py

@@ -11,13 +11,15 @@
 import logging
 
 from proxy.common.utils import socket_connection
-from proxy.common.constants import DEFAULT_BUFFER_SIZE
+from proxy.common.constants import DEFAULT_LOG_FORMAT, DEFAULT_BUFFER_SIZE
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 logger = logging.getLogger(__name__)
 
 if __name__ == '__main__':
-    with socket_connection(('::', 12345)) as client:
+    with socket_connection(('127.0.0.1', 12345)) as client:
         while True:
             client.send(b'hello')
             data = client.recv(DEFAULT_BUFFER_SIZE)

examples/websocket_client.py

@@ -14,8 +14,11 @@
 from proxy.http.websocket import (
     WebsocketFrame, WebsocketClient, websocketOpcodes,
 )
+from proxy.common.constants import DEFAULT_LOG_FORMAT
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 # globals
 client: WebsocketClient
 last_dispatch_time: float
@@ -47,9 +50,9 @@ def on_message(frame: WebsocketFrame) -> None:
 if __name__ == '__main__':
     # Constructor establishes socket connection
     client = WebsocketClient(
-        b'echo.websocket.org',
-        80,
-        b'/',
+        b'localhost',
+        8899,
+        b'/ws-route-example',
         on_message=on_message,
     )
     # Perform handshake

proxy/common/constants.py

@@ -149,6 +149,7 @@ def _env_threadless_compliant() -> bool:
     DEFAULT_DATA_DIRECTORY_PATH, 'cache',
 )
 DEFAULT_CACHE_REQUESTS = False
+DEFAULT_CACHE_BY_CONTENT_TYPE = False
 
 # Cor plugins enabled by default or via flags
 DEFAULT_ABC_PLUGINS = [

proxy/common/flag.py

@@ -398,7 +398,8 @@ def initialize(
         # FIXME: Necessary here until flags framework provides a way
         # for flag owners to initialize
         os.makedirs(args.cache_dir, exist_ok=True)
-        os.makedirs(os.path.join(args.cache_dir, 'response'), exist_ok=True)
+        os.makedirs(os.path.join(args.cache_dir, 'responses'), exist_ok=True)
+        os.makedirs(os.path.join(args.cache_dir, 'content'), exist_ok=True)
 
         return args
 

proxy/common/utils.py

@@ -120,6 +120,7 @@ def build_http_response(
     headers: Optional[Dict[bytes, bytes]] = None,
     body: Optional[bytes] = None,
     conn_close: bool = False,
+    no_cl: bool = False,
 ) -> bytes:
     """Build and returns a HTTP response packet."""
     line = [protocol_version, bytes_(status_code)]
@@ -131,7 +132,7 @@ def build_http_response(
         if k.lower() == b'transfer-encoding':
             has_transfer_encoding = True
             break
-    if not has_transfer_encoding:
+    if not has_transfer_encoding and not no_cl:
         headers[b'Content-Length'] = bytes_(len(body)) if body else b'0'
     return build_http_pkt(line, headers, body, conn_close)
 
@@ -212,12 +213,15 @@ def find_http_line(raw: bytes) -> Tuple[Optional[bytes], bytes]:
 
 
 def wrap_socket(
-    conn: socket.socket, keyfile: str,
-    certfile: str,
+        conn: socket.socket,
+        keyfile: str,
+        certfile: str,
+        cafile: Optional[str] = None,
 ) -> ssl.SSLSocket:
     """Use this to upgrade server_side socket to TLS."""
     ctx = ssl.create_default_context(
         ssl.Purpose.CLIENT_AUTH,
+        cafile=cafile,
     )
     ctx.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
     ctx.verify_mode = ssl.CERT_NONE

proxy/core/base/tcp_server.py

@@ -74,7 +74,7 @@
     type=int,
     default=DEFAULT_MAX_SEND_SIZE,
     help='Default: ' + str(int(DEFAULT_MAX_SEND_SIZE / 1024)) +
-    ' KB. Maximum amount of data to dispatch in a single send() operation.',
+    ' KB. Maximum amount of data to flush in a single send() operation.',
 )
 
 flags.add_argument(
@@ -185,8 +185,19 @@ async def handle_readables(self, readables: Readables) -> bool:
         if self.work.connection.fileno() in readables:
             try:
                 data = self.work.recv(self.flags.client_recvbuf_size)
+            except ConnectionResetError:
+                logger.info(
+                    'Connection reset by client {0}'.format(
+                        self.work.address,
+                    ),
+                )
+                return True
             except TimeoutError:
-                logger.info('Client recv timeout error')
+                logger.info(
+                    'Client recv timeout error {0}'.format(
+                        self.work.address,
+                    ),
+                )
                 return True
             if data is None:
                 logger.debug(

proxy/core/connection/connection.py

@@ -49,7 +49,7 @@ def connection(self) -> TcpOrTlsSocket:
 
     def send(self, data: Union[memoryview, bytes]) -> int:
         """Users must handle BrokenPipeError exceptions"""
-        # logger.info(data)
+        # logger.info(data.tobytes())
         return self.connection.send(data)
 
     def recv(

proxy/core/connection/server.py

@@ -45,15 +45,19 @@ def connect(
 
     def wrap(
             self,
-            hostname: str,
+            hostname: Optional[str] = None,
             ca_file: Optional[str] = None,
             as_non_blocking: bool = False,
+            # Ref https://github.com/PyCQA/pylint/issues/3691
+            verify_mode: ssl.VerifyMode = ssl.VerifyMode.CERT_REQUIRED,     # pylint: disable=E1101
     ) -> None:
         ctx = ssl.create_default_context(
-            ssl.Purpose.SERVER_AUTH, cafile=ca_file,
+            ssl.Purpose.SERVER_AUTH,
+            cafile=ca_file,
         )
         ctx.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
-        ctx.check_hostname = True
+        ctx.check_hostname = hostname is not None
+        ctx.verify_mode = verify_mode
         self.connection.setblocking(True)
         self._conn = ctx.wrap_socket(
             self.connection,