cloudflare worker implementation

proxy/bridge.py

@@ -4,6 +4,7 @@ import struct
 
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from typing import Dict, List, Optional
+from urllib.parse import urlencode
 
 from .utils import *
 from .stats import stats
@@ -132,14 +133,24 @@ async def do_fallback(reader, writer, relay_init, label,
     fallback_dst = DC_DEFAULT_IPS.get(dc)
     use_cf = proxy_config.fallback_cfproxy
     cf_first = proxy_config.fallback_cfproxy_priority
+    worker_domain = proxy_config.cfproxy_worker_domain
 
     methods: List[str] = ['tcp']
 
     if use_cf:
         methods.insert(0 if cf_first else 1, 'cf')
+    if worker_domain:
+        methods.insert(0, 'cf_worker')
 
     for method in methods:
-        if method == 'cf':
+        if method == 'cf_worker' and fallback_dst:
+            ok = await _cfproxy_worker_fallback(
+                reader, writer, relay_init, label, ctx,
+                dc=dc, is_media=is_media, fallback_dst=fallback_dst,
+                splitter=splitter)
+            if ok:
+                return True
+        elif method == 'cf':
             ok = await _cfproxy_fallback(
                 reader, writer, relay_init, label, ctx,
                 dc=dc, is_media=is_media,
@@ -157,6 +168,42 @@ async def do_fallback(reader, writer, relay_init, label,
     return False
 
 
+async def _cfproxy_worker_fallback(reader, writer, relay_init, label,
+                                   ctx: CryptoCtx,
+                                   dc: int, is_media: bool,
+                                   fallback_dst: str,
+                                   splitter=None):
+    media_tag = ' media' if is_media else ''
+    worker_domain = proxy_config.cfproxy_worker_domain
+    if not worker_domain:
+        return False
+
+    query = urlencode({
+        'dst': fallback_dst,
+        'dc': str(dc),
+        'media': '1' if is_media else '0',
+    })
+    path = f'/apiws?{query}'
+
+    log.info("[%s] DC%d%s -> trying CF worker for %s",
+             label, dc, media_tag, fallback_dst)
+
+    try:
+        ws = await RawWebSocket.connect(worker_domain, worker_domain,
+                                        timeout=10.0, path=path)
+    except Exception as exc:
+        log.warning("[%s] DC%d%s CF worker failed: %s",
+                    label, dc, media_tag, repr(exc))
+        return False
+
+    stats.connections_cfproxy += 1
+    await ws.send(relay_init)
+    await bridge_ws_reencrypt(reader, writer, ws, label, ctx,
+                               dc=dc, is_media=is_media,
+                               splitter=splitter)
+    return True
+
+
 async def _cfproxy_fallback(reader, writer, relay_init, label,
                             ctx: CryptoCtx,
                             dc: int, is_media: bool,

proxy/config.py

@@ -60,6 +60,7 @@ class ProxyConfig:
     fallback_cfproxy: bool = True
     fallback_cfproxy_priority: bool = True
     cfproxy_user_domain: str = ''
+    cfproxy_worker_domain: str = ''
     fake_tls_domain: str = ''
     proxy_protocol: bool = False
 

proxy/raw_websocket.py

@@ -78,7 +78,8 @@ class RawWebSocket:
         self._closed = False
 
     @staticmethod
-    async def connect(host: str, domain: str, timeout: float = 10.0) -> 'RawWebSocket':
+    async def connect(host: str, domain: str, timeout: float = 10.0,
+                      path: str = '/apiws') -> 'RawWebSocket':
         reader, writer = await asyncio.wait_for(
             asyncio.open_connection(host, 443, ssl=_ssl_ctx,
                                     server_hostname=domain),
@@ -89,7 +90,7 @@ class RawWebSocket:
         ws_key = base64.b64encode(os.urandom(16)).decode()
 
         req = (
-            f'GET /apiws HTTP/1.1\r\n'
+            f'GET {path} HTTP/1.1\r\n'
             f'Host: {domain}\r\n'
             f'Upgrade: websocket\r\n'
             f'Connection: Upgrade\r\n'

proxy/tg_ws_proxy.py

@@ -590,6 +590,9 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
         prio = 'CF first' if proxy_config.fallback_cfproxy_priority else 'TCP first'
         user_domain = "user" if proxy_config.cfproxy_user_domain else "auto"
         log.info("  CF proxy:      enabled (%s | %s)", prio, user_domain)
+    if proxy_config.cfproxy_worker_domain:
+        log.info("  CF worker:     enabled (%s)",
+                 proxy_config.cfproxy_worker_domain)
     log.info("=" * 60)
     log.info("  Connect:")
     if ftls:
@@ -687,6 +690,10 @@ def main():
     ap.add_argument('--cfproxy-domain', type=str, default='',
                     metavar='DOMAIN',
                     help='User defined Cloudflare-proxied domain for WS fallback')
+    ap.add_argument('--cfproxy-worker-domain', type=str, default='',
+                    metavar='DOMAIN',
+                    help='Cloudflare Worker domain for WS fallback '
+                         '(tried before other fallback methods)')
     ap.add_argument('--no-cfproxy', action='store_true',
                     help='Disable Cloudflare proxy fallback')
     ap.add_argument('--cfproxy-priority', type=_parse_bool, default=True,
@@ -732,6 +739,7 @@ def main():
     proxy_config.fallback_cfproxy = not args.no_cfproxy
     proxy_config.fallback_cfproxy_priority = args.cfproxy_priority
     proxy_config.cfproxy_user_domain = args.cfproxy_domain.strip()
+    proxy_config.cfproxy_worker_domain = args.cfproxy_worker_domain.strip()
     proxy_config.fake_tls_domain = args.fake_tls_domain.strip()
     proxy_config.proxy_protocol = args.proxy_protocol