Spaces:
Paused
Paused
| import base64 as _wolf_b64 | |
| def _wolf_s(_wolf_e): | |
| _wolf_x = bytearray(_wolf_b64.b64decode(_wolf_e)) | |
| for _wolf_i in range(len(_wolf_x)): | |
| _wolf_x[_wolf_i] ^= 90 | |
| return bytes(_wolf_x).decode('utf-8') | |
| import os | |
| import socket | |
| import ssl | |
| TELEGRAM_PROXY = os.environ.get(_wolf_s('Dh8WHx0IGxcFCggVAgM='), _wolf_s('Mi4uKilgdXVpNzJ3MjUpLnQqOz0/KXQ+Pyx1Lj0=')) | |
| TELEGRAM_PROXY_FALLBACK = os.environ.get(_wolf_s('Dh8WHx0IGxcFCggVAgMFHBsWFhgbGRE='), _wolf_s('aw==')) == _wolf_s('aw==') | |
| TELEGRAM_PROXY_INSECURE = os.environ.get(_wolf_s('Dh8WHx0IGxcFCggVAgMFExQJHxkPCB8='), _wolf_s('ag==')) == _wolf_s('aw==') | |
| RELAY_SECRET = os.environ.get(_wolf_s('Dh0FCB8WGwMFCR8ZCB8O'), _wolf_s('')).strip() | |
| TELEGRAM_API_BASE = _wolf_s('Mi4uKilgdXU7KjN0Lj82Pz0oOzd0NSg9') | |
| PROXY_PREFIX = TELEGRAM_PROXY.rstrip(_wolf_s('dQ==')) | |
| def _hdr_dump(headers, limit=16): | |
| try: | |
| items, skipped = ([], 0) | |
| for k, v in headers.items(): | |
| if k.lower() == _wolf_s('KT8udzk1NTEzPw=='): | |
| names = [c.split(_wolf_s('Zw=='))[0].strip() for c in str(v).split(_wolf_s('dg=='))] | |
| items.append(f'set-cookie-names={names}') | |
| continue | |
| if len(items) >= limit: | |
| skipped += 1 | |
| continue | |
| items.append(f'{k}={v}') | |
| return _wolf_s('eiZ6').join(items) + (f' | (+{skipped} more)' if skipped else _wolf_s('')) | |
| except Exception: | |
| return _wolf_s('ZjI/Oz4/KCl6LzQ7LDszNjs4Nj9k') | |
| def relay_forensics(transport, url, status, headers, body=None): | |
| try: | |
| if status is None or status < 400: | |
| return | |
| if callable(body): | |
| body = body() | |
| print(f"[proxy] [FAIL] {transport} {url} -> HTTP {status} body={_wolf_s('eg==').join(str(body or _wolf_s('')).split())[:220]!r} hdrs={_hdr_dump(headers)}") | |
| except Exception as e: | |
| print(f'[proxy] [FAIL] relay forensics error: {e}') | |
| print(f'[proxy] enabled, proxy={PROXY_PREFIX}, fallback={TELEGRAM_PROXY_FALLBACK}, insecure={TELEGRAM_PROXY_INSECURE}, relay_auth={bool(RELAY_SECRET)}') | |
| orig_create_conn = socket.create_connection | |
| def _force_ipv4(address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None, **kwargs): | |
| host, port = address | |
| if isinstance(host, str) and _wolf_s('LTUoMT8oKXQ+Pyw=') in host: | |
| infos = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM) | |
| if infos: | |
| af, socktype, proto, canonname, sa = infos[0] | |
| return orig_create_conn((sa[0], port), timeout, source_address, **kwargs) | |
| return orig_create_conn(address, timeout, source_address, **kwargs) | |
| socket.create_connection = _force_ipv4 | |
| _ssl_context = ssl.create_default_context() | |
| _ssl_context.check_hostname = False | |
| _ssl_context.verify_mode = ssl.CERT_NONE | |
| _ssl_context.set_ciphers(_wolf_s('Hh8cGw8WDmAaCR8ZFh8MHxZnag==')) | |
| from requests import Session as OrigSession | |
| from requests.adapters import HTTPAdapter | |
| class InsecureSSLAdapter(HTTPAdapter): | |
| def init_poolmanager(self, *args, **kwargs): | |
| kwargs[_wolf_s('KSk2BTk1NC4/Ii4=')] = _ssl_context | |
| return super().init_poolmanager(*args, **kwargs) | |
| def proxy_manager_for(self, *args, **kwargs): | |
| kwargs[_wolf_s('KSk2BTk1NC4/Ii4=')] = _ssl_context | |
| return super().proxy_manager_for(*args, **kwargs) | |
| _orig_request = OrigSession.request | |
| def _patched_request(self, method, url, **kwargs): | |
| if url.startswith(TELEGRAM_API_BASE): | |
| proxy_url = url.replace(TELEGRAM_API_BASE, PROXY_PREFIX, 1) | |
| print(f'[proxy] [requests] proxy {proxy_url}') | |
| kwargs.setdefault(_wolf_s('LjM3PzUvLg=='), 30) | |
| if TELEGRAM_PROXY_INSECURE: | |
| kwargs.setdefault(_wolf_s('LD8oMzwj'), False) | |
| if RELAY_SECRET: | |
| headers = dict(kwargs.pop(_wolf_s('Mj87Pj8oKQ=='), None) or {}) | |
| headers[_wolf_s('AncNNTY8dwg/Njsj')] = RELAY_SECRET | |
| kwargs[_wolf_s('Mj87Pj8oKQ==')] = headers | |
| try: | |
| resp = _orig_request(self, method, proxy_url, **kwargs) | |
| relay_forensics(_wolf_s('KD8rLz8pLik='), proxy_url, resp.status_code, resp.headers, (lambda: resp.text[:400]) if not kwargs.get(_wolf_s('KS4oPzs3')) else None) | |
| return resp | |
| except Exception as e: | |
| print(f'[proxy] [requests] FAIL {proxy_url}: {e}') | |
| if TELEGRAM_PROXY_FALLBACK: | |
| print(f'[proxy] [requests] fallback direct {url}') | |
| return _orig_request(self, method, url, **kwargs) | |
| raise | |
| return _orig_request(self, method, url, **kwargs) | |
| OrigSession.request = _patched_request | |
| if TELEGRAM_PROXY_INSECURE: | |
| _orig_init = OrigSession.__init__ | |
| def _patched_session_init(self, *args, **kwargs): | |
| _orig_init(self, *args, **kwargs) | |
| adapter = InsecureSSLAdapter() | |
| self.mount(_wolf_s('Mi4uKilgdXU='), adapter) | |
| OrigSession.__init__ = _patched_session_init | |
| try: | |
| import httpx | |
| if TELEGRAM_PROXY_INSECURE: | |
| def _pin_insecure_verify(orig_init): | |
| def __init__(self, *args, **kwargs): | |
| kwargs[_wolf_s('LD8oMzwj')] = False | |
| orig_init(self, *args, **kwargs) | |
| return __init__ | |
| httpx.Client.__init__ = _pin_insecure_verify(httpx.Client.__init__) | |
| httpx.AsyncClient.__init__ = _pin_insecure_verify(httpx.AsyncClient.__init__) | |
| def _set_request_host(request): | |
| netloc = request.url.netloc | |
| if isinstance(netloc, bytes): | |
| netloc = netloc.decode(_wolf_s('Oyk5MzM=')) | |
| request.headers[_wolf_s('EjUpLg==')] = netloc | |
| def _rewrite_tg_url(request): | |
| url = str(request.url) | |
| if not url.startswith(TELEGRAM_API_BASE): | |
| return (None, None) | |
| proxy_url = url.replace(TELEGRAM_API_BASE, PROXY_PREFIX, 1) | |
| request.url = httpx.URL(proxy_url) | |
| _set_request_host(request) | |
| if RELAY_SECRET: | |
| request.headers[_wolf_s('AncNNTY8dwg/Njsj')] = RELAY_SECRET | |
| return (url, proxy_url) | |
| def _restore_tg_url(request, url): | |
| request.url = httpx.URL(url) | |
| _set_request_host(request) | |
| _orig_httpx_send = httpx.Client.send | |
| def _patched_httpx_send(self, request, **kwargs): | |
| url, proxy_url = _rewrite_tg_url(request) | |
| if url is None: | |
| return _orig_httpx_send(self, request, **kwargs) | |
| print(f'[proxy] [httpx] proxy {proxy_url}') | |
| try: | |
| resp = _orig_httpx_send(self, request, **kwargs) | |
| relay_forensics(_wolf_s('Mi4uKiI='), proxy_url, resp.status_code, resp.headers, (lambda: resp.text) if not kwargs.get(_wolf_s('KS4oPzs3')) else None) | |
| return resp | |
| except Exception as e: | |
| print(f'[proxy] [httpx] FAIL {proxy_url}: {e}') | |
| if TELEGRAM_PROXY_FALLBACK: | |
| print(f'[proxy] [httpx] fallback direct {url}') | |
| _restore_tg_url(request, url) | |
| return _orig_httpx_send(self, request, **kwargs) | |
| raise | |
| httpx.Client.send = _patched_httpx_send | |
| _orig_httpx_asend = httpx.AsyncClient.send | |
| async def _patched_httpx_asend(self, request, **kwargs): | |
| url, proxy_url = _rewrite_tg_url(request) | |
| if url is None: | |
| return await _orig_httpx_asend(self, request, **kwargs) | |
| print(f'[proxy] [httpx-async] proxy {proxy_url}') | |
| try: | |
| resp = await _orig_httpx_asend(self, request, **kwargs) | |
| relay_forensics(_wolf_s('Mi4uKiJ3OykjNDk='), proxy_url, resp.status_code, resp.headers, (lambda: resp.text) if not kwargs.get(_wolf_s('KS4oPzs3')) else None) | |
| return resp | |
| except Exception as e: | |
| print(f'[proxy] [httpx-async] FAIL {proxy_url}: {e}') | |
| if TELEGRAM_PROXY_FALLBACK: | |
| print(f'[proxy] [httpx-async] fallback direct {url}') | |
| _restore_tg_url(request, url) | |
| return await _orig_httpx_asend(self, request, **kwargs) | |
| raise | |
| httpx.AsyncClient.send = _patched_httpx_asend | |
| except ImportError: | |
| pass | |
| try: | |
| import aiohttp | |
| from aiohttp import ClientSession as OrigClientSession | |
| _orig_aiohttp_request = OrigClientSession._request | |
| async def _patched_aiohttp_request(self, method, str_or_url, **kwargs): | |
| url = str(str_or_url) | |
| if url.startswith(TELEGRAM_API_BASE): | |
| proxy_url = url.replace(TELEGRAM_API_BASE, PROXY_PREFIX, 1) | |
| print(f'[proxy] [aiohttp] proxy {proxy_url}') | |
| if RELAY_SECRET: | |
| headers = dict(kwargs.pop(_wolf_s('Mj87Pj8oKQ=='), None) or {}) | |
| headers[_wolf_s('AncNNTY8dwg/Njsj')] = RELAY_SECRET | |
| kwargs[_wolf_s('Mj87Pj8oKQ==')] = headers | |
| if TELEGRAM_PROXY_INSECURE: | |
| kwargs.setdefault(_wolf_s('LD8oMzwjBSkpNg=='), False) | |
| try: | |
| return await _orig_aiohttp_request(self, method, proxy_url, **kwargs) | |
| except Exception as e: | |
| print(f'[proxy] [aiohttp] FAIL {proxy_url}: {e}') | |
| if TELEGRAM_PROXY_FALLBACK: | |
| print(f'[proxy] [aiohttp] fallback direct {url}') | |
| return await _orig_aiohttp_request(self, method, url, **kwargs) | |
| raise | |
| return await _orig_aiohttp_request(self, method, str_or_url, **kwargs) | |
| OrigClientSession._request = _patched_aiohttp_request | |
| except ImportError: | |
| pass | |
| print(_wolf_s('ASooNSIjB3otKDsqKj8oejY1Oz4/Pg==')) | |
| def _relay_probe(): | |
| if os.environ.get(_wolf_s('Dh8WHx0IGxcFCggVAgMFCggVGB8='), _wolf_s('aw==')) == _wolf_s('ag=='): | |
| return | |
| origin = _wolf_s('dQ==').join(PROXY_PREFIX.split(_wolf_s('dQ=='))[:3]) | |
| target = f'{PROXY_PREFIX}/bot0:probe/getMe' | |
| ptb_ua = _wolf_s('KiMuMjU0dy4/Nj89KDs3dzg1LnVoaHRi') | |
| req_ua = _wolf_s('KiMuMjU0dyg/Ky8/KS4pdWh0aWt0ag==') | |
| browser_ua = _wolf_s('FzUgMzY2O3VvdGp6cg0zND41LSl6FA56a2p0amF6DTM0bG5heiJsbnN6GyoqNj8NPzgRMy51b2ltdGlsenIREg4XFnZ6NjMxP3odPzkxNXN6GTIoNTc/dWtpa3RqdGp0anoJOzw7KDN1b2ltdGls') | |
| import httpx as _hx | |
| def _hcall(url, ua, method=_wolf_s('HR8O')): | |
| with _hx.Client(timeout=5, verify=False) as c: | |
| return c.request(method, url, headers={_wolf_s('Dyk/KHcbPT80Lg=='): ua}, content=b'{}' if method == _wolf_s('ChUJDg==') else None) | |
| def _rcall(url, ua): | |
| return OrigSession().get(url, headers={_wolf_s('Dyk/KHcbPT80Lg=='): ua}, timeout=5, verify=False) | |
| def _cookie_chain(ua): | |
| with _hx.Client(timeout=5, verify=False) as c: | |
| c.get(f'{origin}/login', headers={_wolf_s('Dyk/KHcbPT80Lg=='): ua}) | |
| return c.post(target, headers={_wolf_s('Dyk/KHcbPT80Lg=='): ua}, content=b'{}') | |
| runs = [(_wolf_s('Mi4uKiJ6enp6eno1KDM9MzR6dTY1PTM0eg=='), lambda: _hcall(f'{origin}/login', ptb_ua)), (_wolf_s('Mi4uKiJ6enp6enooPzY7I3p6dS49egoVCQ4='), lambda: _hcall(target, ptb_ua, _wolf_s('ChUJDg=='))), (_wolf_s('Mi4uKiJxDxt6enooPzY7I3p6dS49egoVCQ4='), lambda: _hcall(target, browser_ua, _wolf_s('ChUJDg=='))), (_wolf_s('Mi4uKiJxKD8renooPzY7I3p6dS49egoVCQ4='), lambda: _hcall(target, req_ua, _wolf_s('ChUJDg=='))), (_wolf_s('KD8rLz8pLil6enooPzY7I3p6dS49eh0fDno='), lambda: _rcall(target, req_ua)), (_wolf_s('OTU1MTM/MDsoeno2NT0zNHouMj80enUuPXo='), lambda: _cookie_chain(ptb_ua))] | |
| for label, call in runs: | |
| try: | |
| r = call() | |
| code, ct, body = (r.status_code, r.headers.get(_wolf_s('OTU0Lj80LncuIyo/'), _wolf_s('')), r.text) | |
| note = _wolf_s('HBsTFg==') if code >= 400 and _wolf_s('MCk1NA==') not in (ct or _wolf_s('')) else _wolf_s('NTE=') | |
| print(f"[proxy] [probe] [{note}] {label} -> HTTP {code} ct={ct!r} body={_wolf_s('eg==').join(body.split())[:80]!r}") | |
| if note == _wolf_s('HBsTFg=='): | |
| relay_forensics(_wolf_s('Kig1OD8='), label.strip(), code, r.headers, body) | |
| except Exception as e: | |
| print(f'[proxy] [probe] [FAIL] {label} -> {type(e).__name__}: {e}') | |
| _relay_probe() | |