[Multiport] Prepare for listening on multiple ports (#1031)

* Convert `--port` to list of integers

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

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

* Prepare for multiport listener

* Multi listeners, but will fail in theory as no port override is currently performed

* Separate `listener` module

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

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

* Update flags in readme

* Fix imports in listener test

* Fix flag parsing for `port` and `ports`

* Fix tests

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

Author: abhinavsingh

README.md

@@ -2251,8 +2251,8 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
           [--tunnel-remote-port TUNNEL_REMOTE_PORT] [--enable-events]
           [--threadless] [--threaded] [--num-workers NUM_WORKERS]
           [--local-executor LOCAL_EXECUTOR] [--backlog BACKLOG]
-          [--hostname HOSTNAME] [--port PORT] [--port-file PORT_FILE]
-          [--unix-socket-path UNIX_SOCKET_PATH]
+          [--hostname HOSTNAME] [--port PORT] [--ports PORTS [PORTS ...]]
+          [--port-file PORT_FILE] [--unix-socket-path UNIX_SOCKET_PATH]
           [--num-acceptors NUM_ACCEPTORS] [--version] [--log-level LOG_LEVEL]
           [--log-file LOG_FILE] [--log-format LOG_FORMAT]
           [--open-file-limit OPEN_FILE_LIMIT]
@@ -2270,13 +2270,14 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
           [--proxy-pool PROXY_POOL] [--enable-web-server]
           [--enable-static-server] [--static-server-dir STATIC_SERVER_DIR]
           [--min-compression-length MIN_COMPRESSION_LENGTH]
-          [--pac-file PAC_FILE] [--pac-file-url-path PAC_FILE_URL_PATH]
+          [--enable-reverse-proxy] [--pac-file PAC_FILE]
+          [--pac-file-url-path PAC_FILE_URL_PATH]
           [--cloudflare-dns-mode CLOUDFLARE_DNS_MODE]
           [--filtered-upstream-hosts FILTERED_UPSTREAM_HOSTS]
           [--filtered-client-ips FILTERED_CLIENT_IPS]
           [--filtered-url-regex-config FILTERED_URL_REGEX_CONFIG]
 
-proxy.py v2.4.0rc7.dev28+gfbd7b46.d20220120
+proxy.py v2.4.0rc8.dev7+g1871027
 
 options:
   -h, --help            show this help message and exit
@@ -2315,9 +2316,12 @@ options:
                         executors, instead of using underlying OS kernel
                         scheduling algorithm.
   --backlog BACKLOG     Default: 100. Maximum number of pending connections to
-                        proxy server
+                        proxy server.
   --hostname HOSTNAME   Default: 127.0.0.1. Server IP address.
-  --port PORT           Default: 8899. Server port.
+  --port PORT           Default: 8899. Server port. To listen on more ports,
+                        pass them using --ports flag.
+  --ports PORTS [PORTS ...]
+                        Default: None. Additional ports to listen on.
   --port-file PORT_FILE
                         Default: None. Save server port numbers. Useful when
                         using --port=0 ephemeral mode.
@@ -2420,6 +2424,8 @@ options:
   --min-compression-length MIN_COMPRESSION_LENGTH
                         Default: 20 bytes. Sets the minimum length of a
                         response that will be compressed (gzipped).
+  --enable-reverse-proxy
+                        Default: False. Whether to enable reverse proxy core.
   --pac-file PAC_FILE   A file (Proxy Auto Configuration) or string to serve
                         when the server receives a direct file request. Using
                         this option enables proxy.HttpWebServerPlugin.

proxy/common/flag.py

@@ -302,6 +302,7 @@ def initialize(
             # assert args.unix_socket_path is None
             args.family = socket.AF_INET6 if args.hostname.version == 6 else socket.AF_INET
         args.port = cast(int, opts.get('port', args.port))
+        args.ports = cast(Optional[List[int]], opts.get('ports', args.ports))
         args.backlog = cast(int, opts.get('backlog', args.backlog))
         num_workers = opts.get('num_workers', args.num_workers)
         args.num_workers = cast(

proxy/core/acceptor/__init__.py

@@ -14,11 +14,9 @@
 """
 from .pool import AcceptorPool
 from .acceptor import Acceptor
-from .listener import Listener
 
 
 __all__ = [
-    'Listener',
     'Acceptor',
     'AcceptorPool',
 ]

proxy/core/acceptor/listener.py

@@ -22,7 +22,7 @@
 from multiprocessing.reduction import send_handle
 
 from .acceptor import Acceptor
-from .listener import Listener
+from ..listener import ListenerPool
 from ...common.flag import flags
 from ...common.constants import DEFAULT_NUM_ACCEPTORS
 
@@ -64,15 +64,15 @@ class AcceptorPool:
     def __init__(
             self,
             flags: argparse.Namespace,
-            listener: Listener,
+            listeners: ListenerPool,
             executor_queues: List[connection.Connection],
             executor_pids: List[int],
             executor_locks: List['multiprocessing.synchronize.Lock'],
             event_queue: Optional['EventQueue'] = None,
     ) -> None:
         self.flags = flags
         # File descriptor to use for accepting new work
-        self.listener: Listener = listener
+        self.listeners: ListenerPool = listeners
         # Available executors
         self.executor_queues: List[connection.Connection] = executor_queues
         self.executor_pids: List[int] = executor_pids
@@ -109,14 +109,16 @@ def setup(self) -> None:
             ),
         )
         # Send file descriptor to all acceptor processes.
-        fd = self.listener.fileno()
-        for index in range(self.flags.num_acceptors):
-            send_handle(
-                self.fd_queues[index],
-                fd,
-                self.acceptors[index].pid,
-            )
-            self.fd_queues[index].close()
+        for listener in self.listeners.pool:
+            fd = listener.fileno()
+            assert fd is not None
+            for index in range(self.flags.num_acceptors):
+                send_handle(
+                    self.fd_queues[index],
+                    fd,
+                    self.acceptors[index].pid,
+                )
+                self.fd_queues[index].close()
 
     def shutdown(self) -> None:
         logger.info('Shutting down %d acceptors' % self.flags.num_acceptors)

proxy/core/acceptor/pool.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+
+    .. spelling::
+
+       pre
+"""
+from .tcp import TcpSocketListener
+from .pool import ListenerPool
+from .unix import UnixSocketListener
+
+
+__all__ = [
+    'UnixSocketListener',
+    'TcpSocketListener',
+    'ListenerPool',
+]

proxy/core/listener/__init__.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import os
+import socket
+import logging
+import argparse
+from abc import ABC, abstractmethod
+from typing import Any, Optional
+
+from ...common.flag import flags
+from ...common.constants import DEFAULT_BACKLOG
+
+
+flags.add_argument(
+    '--backlog',
+    type=int,
+    default=DEFAULT_BACKLOG,
+    help='Default: 100. Maximum number of pending connections to proxy server.',
+)
+
+logger = logging.getLogger(__name__)
+
+
+class BaseListener(ABC):
+    """Base listener class.
+
+    For usage provide a listen method implementation."""
+
+    def __init__(self, flags: argparse.Namespace) -> None:
+        self.flags = flags
+        self._socket: Optional[socket.socket] = None
+
+    @abstractmethod
+    def listen(self) -> socket.socket:
+        raise NotImplementedError()
+
+    def __enter__(self) -> 'BaseListener':
+        self.setup()
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.shutdown()
+
+    def fileno(self) -> Optional[int]:
+        if not self._socket:
+            return None
+        return self._socket.fileno()
+
+    def setup(self) -> None:
+        self._socket = self.listen()
+
+    def shutdown(self) -> None:
+        assert self._socket
+        self._socket.close()
+        if self.flags.unix_socket_path:
+            os.remove(self.flags.unix_socket_path)

proxy/core/listener/base.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import argparse
+from typing import TYPE_CHECKING, Any, List
+
+from .tcp import TcpSocketListener
+from .unix import UnixSocketListener
+
+
+if TYPE_CHECKING:
+    from .base import BaseListener
+
+
+class ListenerPool:
+    """Provides abstraction around starting multiple listeners
+    based upon flags."""
+
+    def __init__(self, flags: argparse.Namespace) -> None:
+        self.flags = flags
+        self.pool: List['BaseListener'] = []
+
+    def __enter__(self) -> 'ListenerPool':
+        self.setup()
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.shutdown()
+
+    def setup(self) -> None:
+        if self.flags.unix_socket_path:
+            ulistener = UnixSocketListener(flags=self.flags)
+            ulistener.setup()
+            self.pool.append(ulistener)
+        else:
+            listener = TcpSocketListener(flags=self.flags)
+            listener.setup()
+            self.pool.append(listener)
+
+    def shutdown(self) -> None:
+        for listener in self.pool:
+            listener.shutdown()
+        self.pool.clear()