astelm/tg-ws-proxy
1
0
Fork 0
mirror of https://github.com/Flowseal/tg-ws-proxy.git synced 2026-06-11 01:01:43 +03:00
Code Issues Packages Projects Releases Wiki Activity

Possibility to pass few cfproxy and worker domains

Browse Source
This commit is contained in:
Flowseal
2026-05-30 20:30:47 +03:00
parent 05d6de269b
commit ea4e8e790a
10 changed files with 202 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
proxy/__init__.py
View File

@@ -1,6 +1,6 @@
from .config import parse_dc_ip_list, proxy_config
from .config import parse_dc_ip_list, proxy_config, coerce_domain_list
from .utils import get_link_host, build_github_opener
__version__ = "1.7.0"
__all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list", "build_github_opener"]
__all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list", "build_github_opener", "coerce_domain_list"]

60
proxy/bridge.py
View File

@@ -133,11 +133,11 @@ async def do_fallback(reader, writer, relay_init, label,
ctx: CryptoCtx, splitter=None):
fallback_dst = DC_DEFAULT_IPS.get(dc)
use_cf = proxy_config.fallback_cfproxy
worker_domain = proxy_config.cfproxy_worker_domain
worker_domains = proxy_config.cfproxy_worker_domains
methods: List[str] = []
if worker_domain and fallback_dst:
if worker_domains and fallback_dst:
methods.append('cf_worker')
if use_cf:
methods.append('cf')
@@ -176,38 +176,40 @@ async def _cfproxy_worker_fallback(reader, writer, relay_init, label,
fallback_dst: str,
splitter=None):
media_tag = ' media' if is_media else ''
worker_domain = proxy_config.cfproxy_worker_domain
if not worker_domain:
worker_domains = proxy_config.cfproxy_worker_domains
if not worker_domains:
return False
ws = await cf_worker_pool.get(dc, worker_domain, fallback_dst)
if ws:
log.info("[%s] DC%d%s -> CF worker pool hit for %s",
label, dc, media_tag, fallback_dst)
else:
query = urlencode({
'dst': fallback_dst,
'dc': str(dc),
})
path = f'/apiws?{query}'
for worker_domain in worker_domains:
ws = await cf_worker_pool.get(dc, worker_domain, fallback_dst)
if ws:
log.info("[%s] DC%d%s -> CF worker pool hit for %s",
label, dc, media_tag, fallback_dst)
else:
query = urlencode({
'dst': fallback_dst,
'dc': str(dc),
})
path = f'/apiws?{query}'
log.info("[%s] DC%d%s -> trying CF worker for %s",
label, dc, media_tag, fallback_dst)
log.info("[%s] DC%d%s -> trying CF worker %s for %s",
label, dc, media_tag, worker_domain, 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
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 %s failed: %s",
label, dc, media_tag, worker_domain, repr(exc))
continue
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
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
return False
async def _cfproxy_fallback(reader, writer, relay_init, label,

30
proxy/config.py
View File

@@ -58,8 +58,8 @@ class ProxyConfig:
buffer_size: int = 256 * 1024
pool_size: int = 4
fallback_cfproxy: bool = True
cfproxy_user_domain: str = ''
cfproxy_worker_domain: str = ''
cfproxy_user_domains: List[str] = field(default_factory=list)
cfproxy_worker_domains: List[str] = field(default_factory=list)
fake_tls_domain: str = ''
proxy_protocol: bool = False
@@ -67,6 +67,30 @@ class ProxyConfig:
proxy_config = ProxyConfig()
def coerce_domain_list(value) -> List[str]:
if isinstance(value, str):
items = value.replace(',', ' ').replace(';', ' ').split()
elif isinstance(value, (list, tuple)):
items: List[str] = []
for entry in value:
if isinstance(entry, str):
items.extend(entry.replace(',', ' ').replace(';', ' ').split())
else:
return []
seen = set()
result: List[str] = []
for item in items:
item = item.strip()
if not item:
continue
key = item.lower()
if key in seen:
continue
seen.add(key)
result.append(item)
return result
def _fetch_cfproxy_domain_list() -> List[str]:
try:
req = Request(CFPROXY_DOMAINS_URL + "?" + "".join(random.choices(string.ascii_letters, k=7)),
@@ -120,7 +144,7 @@ def _normalize_domain_pool(domains: List[str]) -> List[str]:
def refresh_cfproxy_domains() -> None:
if proxy_config.cfproxy_user_domain:
if proxy_config.cfproxy_user_domains:
return
fetched = _fetch_cfproxy_domain_list()

39
proxy/pool.py
View File

@@ -114,16 +114,17 @@ class _CfWorkerPool:
WS_POOL_MAX_AGE = 120.0
def __init__(self):
self._idle: Dict[int, deque] = {}
self._refilling: Set[int] = set()
self._idle: Dict[Tuple[int, str], deque] = {}
self._refilling: Set[Tuple[int, str]] = set()
async def get(self, dc: int, worker_domain: str, fallback_dst: str) -> Optional[RawWebSocket]:
now = time.monotonic()
key = (dc, worker_domain)
bucket = self._idle.get(dc)
bucket = self._idle.get(key)
if bucket is None:
bucket = deque()
self._idle[dc] = bucket
self._idle[key] = bucket
while bucket:
ws, created = bucket.popleft()
age = now - created
@@ -134,22 +135,23 @@ class _CfWorkerPool:
stats.cf_pool_hits += 1
log.debug("CF worker pool hit DC%d (age=%.1fs, left=%d)",
dc, age, len(bucket))
self._schedule_refill(dc, worker_domain, fallback_dst)
self._schedule_refill(key, fallback_dst)
return ws
stats.cf_pool_misses += 1
self._schedule_refill(dc, worker_domain, fallback_dst)
self._schedule_refill(key, fallback_dst)
return None
def _schedule_refill(self, dc, worker_domain, fallback_dst):
if dc in self._refilling:
def _schedule_refill(self, key, fallback_dst):
if key in self._refilling:
return
self._refilling.add(dc)
asyncio.create_task(self._refill(dc, worker_domain, fallback_dst))
self._refilling.add(key)
asyncio.create_task(self._refill(key, fallback_dst))
async def _refill(self, dc, worker_domain, fallback_dst):
async def _refill(self, key, fallback_dst):
dc, worker_domain = key
try:
bucket = self._idle.setdefault(dc, deque())
bucket = self._idle.setdefault(key, deque())
needed = proxy_config.pool_size - len(bucket)
if needed <= 0:
return
@@ -166,7 +168,7 @@ class _CfWorkerPool:
log.debug("CF worker pool refilled DC%d: %d ready",
dc, len(bucket))
finally:
self._refilling.discard(dc)
self._refilling.discard(key)
@staticmethod
async def _connect_one(worker_domain, fallback_dst, dc) -> Optional[RawWebSocket]:
@@ -194,12 +196,13 @@ class _CfWorkerPool:
if dc not in proxy_config.dc_redirects
}
if not cf_fallbacks or not proxy_config.cfproxy_worker_domain:
if not cf_fallbacks or not proxy_config.cfproxy_worker_domains:
return
for dc, fallback_dst in cf_fallbacks.items():
self._schedule_refill(dc, proxy_config.cfproxy_worker_domain, fallback_dst)
for worker_domain in proxy_config.cfproxy_worker_domains:
for dc, fallback_dst in cf_fallbacks.items():
self._schedule_refill((dc, worker_domain), fallback_dst)
log.info("CF worker pool warmup started for %d DC(s)", len(cf_fallbacks))
def reset(self):

26
proxy/tg_ws_proxy.py
View File

@@ -22,7 +22,7 @@ if __name__ == '__main__' and (__package__ is None or __package__ == ''):
from .utils import *
from .stats import stats
from .config import proxy_config, parse_dc_ip_list, start_cfproxy_domain_refresh
from .config import proxy_config, parse_dc_ip_list, start_cfproxy_domain_refresh, coerce_domain_list
from .bridge import MsgSplitter, CryptoCtx, do_fallback, bridge_ws_reencrypt
from .raw_websocket import RawWebSocket, WsHandshakeError, set_sock_opts
from .fake_tls import proxy_to_masking_domain, verify_client_hello, build_server_hello, FakeTlsStream, TLS_RECORD_HANDSHAKE
@@ -439,9 +439,9 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
_client_tasks.clear()
if proxy_config.fallback_cfproxy:
user = proxy_config.cfproxy_user_domain
user = proxy_config.cfproxy_user_domains
if user:
balancer.update_domains_list([user])
balancer.update_domains_list(user)
else:
start_cfproxy_domain_refresh()
@@ -484,11 +484,11 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
ip = proxy_config.dc_redirects.get(dc)
log.info(" DC%d: %s", dc, ip)
if proxy_config.fallback_cfproxy:
user_domain = "user" if proxy_config.cfproxy_user_domain else "auto"
user_domain = "user" if proxy_config.cfproxy_user_domains else "auto"
log.info(" CF proxy: enabled (%s)", user_domain)
if proxy_config.cfproxy_worker_domain:
if proxy_config.cfproxy_worker_domains:
log.info(" CF worker: enabled (%s)",
proxy_config.cfproxy_worker_domain)
", ".join(proxy_config.cfproxy_worker_domains))
log.info("=" * 60)
log.info(" Connect:")
if ftls:
@@ -574,13 +574,15 @@ def main():
help='Socket send/recv buffer size in KB (default 256)')
ap.add_argument('--pool-size', type=int, default=4, metavar='N',
help='WS connection pool size per DC (default 4, min 0)')
ap.add_argument('--cfproxy-domain', type=str, default='',
ap.add_argument('--cfproxy-domain', action='append', default=None,
metavar='DOMAIN',
help='User defined Cloudflare-proxied domain for WS fallback')
ap.add_argument('--cfproxy-worker-domain', type=str, default='',
help='User defined Cloudflare-proxied domain for WS fallback '
'(repeatable for multiple domains)')
ap.add_argument('--cfproxy-worker-domain', action='append', default=None,
metavar='DOMAIN',
help='Cloudflare Worker domain for WS fallback '
'(tried before other fallback methods)')
'(tried before other fallback methods, '
'repeatable for multiple domains)')
ap.add_argument('--no-cfproxy', action='store_true',
help='Disable Cloudflare proxy fallback')
ap.add_argument('--fake-tls-domain', type=str, default='',
@@ -622,8 +624,8 @@ def main():
proxy_config.buffer_size = max(4, args.buf_kb) * 1024
proxy_config.pool_size = max(0, args.pool_size)
proxy_config.fallback_cfproxy = not args.no_cfproxy
proxy_config.cfproxy_user_domain = args.cfproxy_domain.strip()
proxy_config.cfproxy_worker_domain = args.cfproxy_worker_domain.strip()
proxy_config.cfproxy_user_domains = coerce_domain_list(args.cfproxy_domain)
proxy_config.cfproxy_worker_domains = coerce_domain_list(args.cfproxy_worker_domain)
proxy_config.fake_tls_domain = args.fake_tls_domain.strip()
proxy_config.proxy_protocol = args.proxy_protocol