This application is running normally.
" + + "" + ); +} + +function _json(obj) { + return ContentService.createTextOutput(JSON.stringify(obj)).setMimeType( + ContentService.MimeType.JSON + ); +} diff --git a/README_FA.md b/README_FA.md new file mode 100644 index 0000000..e9e20c9 --- /dev/null +++ b/README_FA.md @@ -0,0 +1,289 @@ +# MasterHttpRelayVPN + +**[English README](README.md)** + +یک ابزار رایگان برای عبور از فیلترینگ و DPI که ترافیک شما را پشت دامنههای قابل اعتماد مثل Google پنهان میکند. برای حالت ساده، به VPS یا سرور نیاز ندارید و فقط یک اکانت Google کافی است. + +> **توضیح ساده:** مرورگر شما به این ابزار روی کامپیوتر خودتان وصل میشود. این ابزار ترافیک را شبیه ترافیک عادی Google نشان میدهد. فیلتر فقط `google.com` را میبیند و اجازه عبور میدهد. در پشت صحنه، یک Google Apps Script رایگان سایت واقعی را برای شما دریافت میکند. + +--- + +## کانال اطلاعرسانی و پشتیبانی 📢 + +برای دریافت آخرین اخبار، نسخهها و اطلاعیههای پروژه، کانال تلگرام ما را دنبال کنید: [Telegram Channel](https://t.me/masterdnsvpn) + +--- + +## نحوه کار + +``` +مرورگر -> پراکسی محلی -> Google/CDN -> رله شما -> سایت مقصد + | + +-> فیلتر فقط google.com را میبیند +``` + +مرورگر، درخواستها را به پراکسی محلی میفرستد. پراکسی این درخواستها را از مسیر Google عبور میدهد تا برای فیلتر شبیه ترافیک عادی به نظر برسد. سپس رلهای که شما deploy کردهاید، سایت اصلی را دریافت میکند و پاسخ را برمیگرداند. + +--- + +## راهاندازی مرحلهبهمرحله + +### مرحله 1: دریافت پروژه + +```bash +git clone -b python_testing https://github.com/masterking32/MasterHttpRelayVPN.git +cd MasterHttpRelayVPN +pip install -r requirements.txt +``` + +> **دسترسی به PyPI ندارید؟** از این mirror استفاده کنید: +> ```bash +> pip install -r requirements.txt -i https://mirror-pypi.runflare.com/simple/ --trusted-host mirror-pypi.runflare.com +> ``` + +اگر نخواستید با Git کار کنید، میتوانید فایل ZIP پروژه را از GitHub دانلود و extract کنید. + +### مرحله 2: راهاندازی رله Google با `Code.gs` + +این بخش همان رلهای است که روی سرورهای Google اجرا میشود و سایتها را برای شما دریافت میکند. + +1. وارد [Google Apps Script](https://script.google.com/) شوید. +2. روی **New project** کلیک کنید. +3. کد پیشفرض را کامل حذف کنید. +4. فایل `Code.gs` همین پروژه را باز کنید، همه محتوای آن را کپی کنید و داخل Apps Script قرار دهید. +5. این خط را به یک رمز دلخواه و امن تغییر دهید: + ```javascript + const AUTH_KEY = "your-secret-password-here"; + ``` +6. روی **Deploy -> New deployment** کلیک کنید. +7. نوع deployment را **Web app** بگذارید. +8. این تنظیمات را انتخاب کنید: + - **Execute as:** Me + - **Who has access:** Anyone +9. روی **Deploy** بزنید. +10. مقدار **Deployment ID** را کپی کنید. در مرحله بعد به آن نیاز دارید. + +نکته: مقداری که برای `AUTH_KEY` میگذارید باید دقیقا با `auth_key` در فایل `config.json` یکی باشد. + +### مرحله 3: تنظیم `config.json` + +ابتدا فایل نمونه را کپی کنید: + +```bash +cp config.example.json config.json +``` + +در ویندوز میتوانید فایل را دستی کپی و rename کنید. + +سپس `config.json` را باز کنید و مقادیر را وارد کنید: + +```json +{ + "mode": "apps_script", + "google_ip": "216.239.38.120", + "front_domain": "www.google.com", + "script_id": "PASTE_YOUR_DEPLOYMENT_ID_HERE", + "auth_key": "your-secret-password-here", + "listen_host": "127.0.0.1", + "listen_port": 8085, + "log_level": "INFO", + "verify_ssl": true +} +``` + +- `script_id` : همان Deployment ID مرحله 2 +- `auth_key` : همان رمزی که در `Code.gs` گذاشتهاید + +### مرحله 4: اجرا + +```bash +python main.py +``` + +اگر همهچیز درست باشد، پراکسی روی `127.0.0.1:8085` بالا میآید. + +### مرحله 5: تنظیم مرورگر + +مرورگر را روی این پراکسی تنظیم کنید: + +- **Proxy Address:** `127.0.0.1` +- **Proxy Port:** `8085` +- **Type:** HTTP + +نمونه تنظیم مرورگرها: + +- **Firefox:** Settings -> General -> Network Settings -> Manual proxy +- **Chrome / Edge:** از تنظیمات پراکسی سیستم استفاده میکنند +- یا از افزونههایی مثل FoxyProxy استفاده کنید + +### مرحله 6: نصب گواهی CA برای HTTPS + +در حالت `apps_script`، برنامه برای مدیریت HTTPS یک گواهی محلی میسازد. اگر آن را نصب نکنید، مرورگر برای سایتها خطای امنیتی میدهد. + +فایل گواهی بعد از اولین اجرا در این مسیر ساخته میشود: + +`ca/ca.crt` + +#### ویندوز +1. روی `ca/ca.crt` دوبار کلیک کنید. +2. گزینه **Install Certificate** را بزنید. +3. گزینه **Current User** را انتخاب کنید. +4. گزینه **Place all certificates in the following store** را بزنید. +5. از بخش **Browse**، گزینه **Trusted Root Certification Authorities** را انتخاب کنید. +6. مراحل را تا پایان ادامه دهید. +7. مرورگر را یک بار ببندید و دوباره باز کنید. + +#### Firefox +Firefox معمولا certificate store جداگانه دارد: + +1. به **Settings -> Privacy & Security -> Certificates** بروید. +2. روی **View Certificates** کلیک کنید. +3. در تب **Authorities**، روی **Import** بزنید. +4. فایل `ca/ca.crt` را انتخاب کنید. +5. گزینه **Trust this CA to identify websites** را فعال کنید. + +> **نصب خودکار هنگام اجرا:** در حالت `apps_script`، برنامه به صورت خودکار وضعیت اعتماد گواهی CA را بررسی کرده و در صورت نیاز نصب میکند. در صورت موفقیت پیام تأیید در لاگ نمایش داده میشود. اگر نصب خودکار ناموفق بود، میتوانید دستور `python main.py --install-cert` را اجرا کنید. + +نکته امنیتی: پوشه `ca/` را با کسی به اشتراک نگذارید. اگر خواستید از اول گواهی جدید بسازید، این پوشه را حذف کنید تا دوباره ساخته شود. + +--- + +## حالتهای موجود + +| حالت | نیازمندی | توضیح | +|------|----------|-------| +| `apps_script` | اکانت رایگان Google | سادهترین حالت، بدون نیاز به سرور | +| `google_fronting` | Google Cloud Run | استفاده از سرویس Cloud Run خودتان | +| `domain_fronting` | Cloudflare Worker | استفاده از Worker روی Cloudflare | +| `custom_domain` | دامنه شخصی روی Cloudflare | اتصال مستقیم به دامنه خودتان | + +برای اکثر کاربران، `apps_script` بهترین انتخاب است. + +--- + +## تنظیمات مهم + +| تنظیم | توضیح | +|------|-------| +| `mode` | نوع رله | +| `auth_key` | رمز مشترک بین برنامه و رله | +| `script_id` | Deployment ID مربوط به Apps Script | +| `listen_host` | آدرس محلی برای اجرا | +| `listen_port` | پورت پراکسی | +| `log_level` | میزان جزئیات لاگ | + +### تنظیمات پیشرفته + +| تنظیم | مقدار پیشفرض | توضیح | +|------|---------------|-------| +| `google_ip` | `216.239.38.120` | IP مورد استفاده برای مسیر Google | +| `front_domain` | `www.google.com` | دامنهای که فیلتر میبیند | +| `verify_ssl` | `true` | بررسی اعتبار TLS | +| `worker_host` | - | برای حالتهای Cloudflare/Cloud Run | +| `custom_domain` | - | دامنه شخصی شما | +| `script_ids` | - | چند Deployment ID برای load balancing | + +### استفاده از چند Script ID + +اگر چند نسخه از `Code.gs` را deploy کنید، میتوانید همه Deployment ID ها را در آرایه `script_ids` بگذارید: + +```json +{ + "script_ids": [ + "DEPLOYMENT_ID_1", + "DEPLOYMENT_ID_2", + "DEPLOYMENT_ID_3" + ] +} +``` + +--- + +## بهروزرسانی `Code.gs` + +اگر فایل `Code.gs` را تغییر دادید، باید دوباره **Deploy -> New deployment** بزنید و `script_id` جدید را داخل `config.json` قرار دهید. صرفا ذخیره کردن کد، نسخه فعال را عوض نمیکند. + +--- + +## دستورهای اجرا + +```bash +python main.py +python main.py -p 9090 +python main.py --log-level DEBUG +python main.py -c /path/to/config.json +python main.py --install-cert # نصب گواهی CA و خروج +python main.py --no-cert-check # رد شدن از بررسی خودکار گواهی +``` + +> **نصب خودکار:** هنگام اجرا در حالت `apps_script`، برنامه بهطور خودکار بررسی میکند که آیا گواهی CA قابل اعتماد است یا نه و در صورت نیاز آن را نصب میکند. اگر نصب خودکار ناموفق بود (مثلاً نیاز به دسترسی مدیر دارد)، میتوانید دستور `python main.py --install-cert` را اجرا کنید یا مراحل مرحله ۶ را دنبال کنید. + +--- + +## معماری + +``` +┌─────────┐ ┌──────────────┐ ┌─────────────┐ ┌──────────┐ +│ Browser │────►│ Local Proxy │────►│ CDN / Google │────►│ Relay │──► Internet +│ │◄────│ (this tool) │◄────│ (fronted) │◄────│ Endpoint │◄── +└─────────┘ └──────────────┘ └─────────────┘ └──────────┘ +``` + +--- + +## فایلهای پروژه + +| فایل | کاربرد | +|------|--------| +| `main.py` | اجرای برنامه | +| `proxy_server.py` | مدیریت اتصال مرورگر | +| `domain_fronter.py` | انجام domain fronting | +| `h2_transport.py` | ارتباط سریعتر با HTTP/2 | +| `mitm.py` | ساخت و مدیریت certificate | +| `cert_installer.py` | نصب خودکار گواهی CA در ویندوز، مک، لینوکس و Firefox | +| `ws.py` | پشتیبانی WebSocket | +| `Code.gs` | رله Apps Script | +| `config.example.json` | فایل نمونه تنظیمات | + +--- + +## رفع مشکل + +| مشکل | راهحل | +|------|--------| +| `Config not found` | فایل `config.example.json` را به `config.json` کپی کنید | +| خطای certificate در مرورگر | گواهی CA را نصب کنید (مرحله ۶) | +| تلگرام کار میکند ولی مرورگر سایتها را باز نمیکند | تقریباً مطمئناً گواهی CA نصب نشده. مرحله ۶ را دنبال کنید، سپس مرورگر را **کاملاً ببندید و دوباره باز کنید** (برای Chrome/Edge مطمئن شوید هیچ پروسه Chrome در پسزمینه باز نیست). | +| گواهی نصب شد ولی مرورگر هنوز خطا میدهد | Chrome و Edge گواهیها را cache میکنند — باید مرورگر را **کاملاً ببندید** (Task Manager یا system tray را چک کنید) و دوباره باز کنید. Firefox نیاز به import جداگانه دارد (بخش Firefox در مرحله ۶). | +| خطای `unauthorized` | مقدار `auth_key` و `AUTH_KEY` باید یکسان باشند | +| timeout | IP دیگری برای Google امتحان کنید | +| سرعت کم | از چند `script_id` برای load balancing استفاده کنید | +| خطای `502 Bad JSON` | Google بهجای JSON پاسخ HTML برگردانده (مثلاً صفحه quota یا 404). دلایل: `script_id` اشتباه، تجاوز از سهمیه روزانه Apps Script، یا عدم ایجاد deployment جدید پس از ویرایش `Code.gs`. `script_id` را بررسی کنید و یک **deployment جدید** بسازید. | + +--- + +## نکات امنیتی + +- فایل `config.json` را با کسی به اشتراک نگذارید. +- مقدار پیشفرض `AUTH_KEY` را قبل از deploy عوض کنید. +- پوشه `ca/` را منتشر نکنید. +- بهتر است `listen_host` روی `127.0.0.1` بماند. + +--- + +## سلب مسئولیت + +MasterHttpRelayVPN فقط برای اهداف آموزشی، تست و پژوهش ارائه شده است. + +- **بدون ضمانت:** این نرمافزار به صورت «همانگونه که هست» ارائه میشود و هیچگونه ضمانت صریح یا ضمنی، از جمله قابلیت فروش، مناسب بودن برای هدف خاص، یا عدم نقض حقوق دیگران برای آن وجود ندارد. +- **محدودیت مسئولیت:** توسعهدهندگان و مشارکتکنندگان این پروژه هیچ مسئولیتی در قبال خسارتهای مستقیم، غیرمستقیم، اتفاقی، تبعی، یا هر نوع خسارت دیگر ناشی از استفاده یا ناتوانی در استفاده از این پروژه ندارند. +- **مسئولیت کاربر:** استفاده از این پروژه خارج از محیطهای کنترلشده و آزمایشی ممکن است بر شبکه، حسابها، پراکسیها، گواهیها یا سیستمهای متصل اثر بگذارد. تمام مسئولیت نصب، پیکربندی و استفاده بر عهده کاربر است. +- **رعایت قوانین:** پیش از استفاده از این نرمافزار، رعایت تمام قوانین و مقررات محلی، کشوری و بینالمللی بر عهده کاربر است. +- **رعایت قوانین Google:** اگر از Google Apps Script یا دیگر سرویسهای Google در این پروژه استفاده میکنید، مسئولیت رعایت شرایط استفاده، محدودیتها، سهمیهها و سیاستهای پلتفرم Google با خود شما است. استفاده نادرست ممکن است باعث تعلیق یا غیرفعال شدن اکانت Google یا deployment های شما شود. +- **شرایط مجوز:** استفاده، کپی، توزیع و تغییر این نرمافزار فقط تحت شرایط مجوز موجود در مخزن مجاز است و هر استفاده خارج از آن شرایط ممنوع است. + +--- + +## License + +MIT diff --git a/cert_installer.py b/cert_installer.py new file mode 100644 index 0000000..1c49762 --- /dev/null +++ b/cert_installer.py @@ -0,0 +1,365 @@ +""" +Cross-platform trusted CA certificate installer. + +Supports: Windows, macOS, Linux (Debian/Ubuntu, RHEL/Fedora/CentOS, Arch). +Also attempts to install into Firefox's NSS certificate store when found. + +Usage: + from cert_installer import install_ca, is_ca_trusted + install_ca("/path/to/ca.crt", cert_name="MasterHttpRelayVPN") +""" + +import glob +import logging +import os +import platform +import shutil +import subprocess +import sys +import tempfile + +log = logging.getLogger("CertInstaller") + +CERT_NAME = "MasterHttpRelayVPN" + + +# ───────────────────────────────────────────────────────────────────────────── +# Helpers +# ───────────────────────────────────────────────────────────────────────────── + +def _run(cmd: list[str], *, check: bool = True, capture: bool = True) -> subprocess.CompletedProcess: + return subprocess.run( + cmd, + check=check, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.PIPE if capture else None, + ) + + +def _has_cmd(name: str) -> bool: + return shutil.which(name) is not None + + +# ───────────────────────────────────────────────────────────────────────────── +# Windows +# ───────────────────────────────────────────────────────────────────────────── + +def _install_windows(cert_path: str, cert_name: str) -> bool: + """ + Install into the current user's Trusted Root store (no admin required). + Falls back to the system store if certutil fails. + """ + # Per-user store — works without elevation + try: + _run(["certutil", "-addstore", "-user", "Root", cert_path]) + log.info("Certificate installed in Windows user Trusted Root store.") + return True + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.warning("certutil user store failed: %s", exc) + + # Try system store (requires admin) + try: + _run(["certutil", "-addstore", "Root", cert_path]) + log.info("Certificate installed in Windows system Trusted Root store.") + return True + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.error("certutil system store failed: %s", exc) + + # Fallback: use PowerShell + try: + ps_cmd = ( + f"Import-Certificate -FilePath '{cert_path}' " + f"-CertStoreLocation Cert:\\CurrentUser\\Root" + ) + _run(["powershell", "-NoProfile", "-Command", ps_cmd]) + log.info("Certificate installed via PowerShell.") + return True + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.error("PowerShell install failed: %s", exc) + + return False + + +def _is_trusted_windows(cert_path: str) -> bool: + # Check by thumbprint only — the cert name is also the folder name, so + # a plain string search on certutil output would produce a false positive. + thumbprint = _cert_thumbprint(cert_path) + if not thumbprint: + return False + try: + result = _run(["certutil", "-user", "-store", "Root"]) + output = result.stdout.decode(errors="replace").upper() + return thumbprint in output + except Exception: + return False + + +def _cert_thumbprint(cert_path: str) -> str: + """Return the SHA-1 thumbprint of a PEM cert (uppercase hex, no colons).""" + try: + from cryptography import x509 as _x509 + from cryptography.hazmat.primitives import hashes as _hashes + with open(cert_path, "rb") as f: + cert = _x509.load_pem_x509_certificate(f.read()) + return cert.fingerprint(_hashes.SHA1()).hex().upper() + except Exception: + return "" + + +# ───────────────────────────────────────────────────────────────────────────── +# macOS +# ───────────────────────────────────────────────────────────────────────────── + +def _install_macos(cert_path: str, cert_name: str) -> bool: + """Install into the login keychain (per-user, no sudo required).""" + login_keychain = os.path.expanduser("~/Library/Keychains/login.keychain-db") + if not os.path.exists(login_keychain): + login_keychain = os.path.expanduser("~/Library/Keychains/login.keychain") + + try: + _run([ + "security", "add-trusted-cert", + "-d", "-r", "trustRoot", + "-k", login_keychain, + cert_path, + ]) + log.info("Certificate installed in macOS login keychain.") + return True + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.warning("login keychain install failed: %s. Trying system keychain (needs sudo)…", exc) + + try: + _run([ + "sudo", "security", "add-trusted-cert", + "-d", "-r", "trustRoot", + "-k", "/Library/Keychains/System.keychain", + cert_path, + ]) + log.info("Certificate installed in macOS system keychain.") + return True + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.error("System keychain install failed: %s", exc) + + return False + + +def _is_trusted_macos(cert_name: str) -> bool: + try: + result = _run(["security", "find-certificate", "-a", "-c", cert_name]) + return bool(result.stdout.strip()) + except Exception: + return False + + +# ───────────────────────────────────────────────────────────────────────────── +# Linux +# ───────────────────────────────────────────────────────────────────────────── + +def _detect_linux_distro() -> str: + """Return 'debian', 'rhel', 'arch', or 'unknown'.""" + if os.path.exists("/etc/debian_version") or os.path.exists("/etc/ubuntu"): + return "debian" + if os.path.exists("/etc/redhat-release") or os.path.exists("/etc/fedora-release"): + return "rhel" + if os.path.exists("/etc/arch-release"): + return "arch" + # Read /etc/os-release as fallback + try: + with open("/etc/os-release") as f: + content = f.read().lower() + if "debian" in content or "ubuntu" in content or "mint" in content: + return "debian" + if "fedora" in content or "rhel" in content or "centos" in content or "rocky" in content or "alma" in content: + return "rhel" + if "arch" in content or "manjaro" in content: + return "arch" + except OSError: + pass + return "unknown" + + +def _install_linux(cert_path: str, cert_name: str) -> bool: + distro = _detect_linux_distro() + log.info("Detected Linux distro family: %s", distro) + + installed = False + + if distro == "debian": + dest_dir = "/usr/local/share/ca-certificates" + dest_file = os.path.join(dest_dir, f"{cert_name.replace(' ', '_')}.crt") + try: + os.makedirs(dest_dir, exist_ok=True) + shutil.copy2(cert_path, dest_file) + _run(["update-ca-certificates"]) + log.info("Certificate installed via update-ca-certificates.") + installed = True + except (OSError, subprocess.CalledProcessError) as exc: + log.warning("Debian install failed (needs sudo?): %s", exc) + # Try with sudo + try: + _run(["sudo", "cp", cert_path, dest_file]) + _run(["sudo", "update-ca-certificates"]) + log.info("Certificate installed via sudo update-ca-certificates.") + installed = True + except (subprocess.CalledProcessError, FileNotFoundError) as exc2: + log.error("sudo Debian install failed: %s", exc2) + + elif distro == "rhel": + dest_dir = "/etc/pki/ca-trust/source/anchors" + dest_file = os.path.join(dest_dir, f"{cert_name.replace(' ', '_')}.crt") + try: + os.makedirs(dest_dir, exist_ok=True) + shutil.copy2(cert_path, dest_file) + _run(["update-ca-trust", "extract"]) + log.info("Certificate installed via update-ca-trust.") + installed = True + except (OSError, subprocess.CalledProcessError) as exc: + log.warning("RHEL install failed (needs sudo?): %s", exc) + try: + _run(["sudo", "cp", cert_path, dest_file]) + _run(["sudo", "update-ca-trust", "extract"]) + log.info("Certificate installed via sudo update-ca-trust.") + installed = True + except (subprocess.CalledProcessError, FileNotFoundError) as exc2: + log.error("sudo RHEL install failed: %s", exc2) + + elif distro == "arch": + dest_dir = "/etc/ca-certificates/trust-source/anchors" + dest_file = os.path.join(dest_dir, f"{cert_name.replace(' ', '_')}.crt") + try: + os.makedirs(dest_dir, exist_ok=True) + shutil.copy2(cert_path, dest_file) + _run(["trust", "extract-compat"]) + log.info("Certificate installed via trust extract-compat.") + installed = True + except (OSError, subprocess.CalledProcessError) as exc: + log.warning("Arch install failed (needs sudo?): %s", exc) + try: + _run(["sudo", "cp", cert_path, dest_file]) + _run(["sudo", "trust", "extract-compat"]) + log.info("Certificate installed via sudo trust extract-compat.") + installed = True + except (subprocess.CalledProcessError, FileNotFoundError) as exc2: + log.error("sudo Arch install failed: %s", exc2) + + else: + log.warning( + "Unknown Linux distro. Manually install %s as a trusted root CA.", cert_path + ) + + return installed + + +def _is_trusted_linux(cert_path: str) -> bool: + """Check if our cert thumbprint is in the system's OpenSSL trust bundle.""" + thumbprint = _cert_thumbprint(cert_path) + if not thumbprint: + return False + bundle_paths = [ + "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu + "/etc/pki/tls/certs/ca-bundle.crt", # RHEL/Fedora + "/etc/ssl/ca-bundle.pem", # OpenSUSE + "/etc/ca-certificates/ca-certificates.crt", + ] + # A fast heuristic: check if our CA cert file was copied to known dirs + anchor_dirs = [ + "/usr/local/share/ca-certificates", + "/etc/pki/ca-trust/source/anchors", + "/etc/ca-certificates/trust-source/anchors", + ] + for d in anchor_dirs: + if os.path.isdir(d): + for f in os.listdir(d): + if "DomainFront" in f or "domainfront" in f.lower(): + return True + return False + + +# ───────────────────────────────────────────────────────────────────────────── +# Firefox NSS (cross-platform) +# ───────────────────────────────────────────────────────────────────────────── + +def _install_firefox(cert_path: str, cert_name: str): + """Install into all detected Firefox profile NSS databases.""" + if not _has_cmd("certutil"): + log.debug("NSS certutil not found — skipping Firefox install.") + return + + profile_dirs: list[str] = [] + system = platform.system() + + if system == "Windows": + appdata = os.environ.get("APPDATA", "") + profile_dirs += glob.glob(os.path.join(appdata, r"Mozilla\Firefox\Profiles\*")) + elif system == "Darwin": + profile_dirs += glob.glob(os.path.expanduser("~/Library/Application Support/Firefox/Profiles/*")) + else: + profile_dirs += glob.glob(os.path.expanduser("~/.mozilla/firefox/*.default*")) + profile_dirs += glob.glob(os.path.expanduser("~/.mozilla/firefox/*.release*")) + + if not profile_dirs: + log.debug("No Firefox profiles found.") + return + + for profile in profile_dirs: + db = f"sql:{profile}" if os.path.exists(os.path.join(profile, "cert9.db")) else f"dbm:{profile}" + try: + # Remove old entry first (ignore errors) + _run(["certutil", "-D", "-n", cert_name, "-d", db], check=False) + _run([ + "certutil", "-A", + "-n", cert_name, + "-t", "CT,,", + "-i", cert_path, + "-d", db, + ]) + log.info("Installed in Firefox profile: %s", os.path.basename(profile)) + except (subprocess.CalledProcessError, FileNotFoundError) as exc: + log.warning("Firefox profile %s: %s", os.path.basename(profile), exc) + + +# ───────────────────────────────────────────────────────────────────────────── +# Public API +# ───────────────────────────────────────────────────────────────────────────── + +def is_ca_trusted(cert_path: str) -> bool: + """Return True if the CA cert appears to be already installed.""" + system = platform.system() + try: + if system == "Windows": + return _is_trusted_windows(cert_path) + if system == "Darwin": + return _is_trusted_macos(CERT_NAME) + return _is_trusted_linux(cert_path) + except Exception: + return False + + +def install_ca(cert_path: str, cert_name: str = CERT_NAME) -> bool: + """ + Install *cert_path* as a trusted root CA on the current platform. + Also attempts Firefox NSS installation. + + Returns True if the system store installation succeeded. + """ + if not os.path.exists(cert_path): + log.error("Certificate file not found: %s", cert_path) + return False + + system = platform.system() + log.info("Installing CA certificate on %s…", system) + + if system == "Windows": + ok = _install_windows(cert_path, cert_name) + elif system == "Darwin": + ok = _install_macos(cert_path, cert_name) + elif system == "Linux": + ok = _install_linux(cert_path, cert_name) + else: + log.error("Unsupported platform: %s", system) + return False + + # Best-effort Firefox install on all platforms + _install_firefox(cert_path, cert_name) + + return ok diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..387a0ed --- /dev/null +++ b/config.example.json @@ -0,0 +1,14 @@ +{ + "_comment": "Copy this file to config.json and fill in your values", + "mode": "apps_script", + "google_ip": "216.239.38.120", + "front_domain": "www.google.com", + "script_id": "YOUR_APPS_SCRIPT_DEPLOYMENT_ID", + "auth_key": "CHANGE_ME_TO_A_STRONG_SECRET", + "listen_host": "127.0.0.1", + "listen_port": 8085, + "log_level": "INFO", + "verify_ssl": true, + "_hosts_comment": "Optional SNI-rewrite overrides. YouTube, googlevideo, gstatic, fonts.googleapis.com, ytimg, ggpht, doubleclick, etc. are ALREADY handled automatically (routed via google_ip with SNI=front_domain, same trick as the Xray MITM-DomainFronting config). Add entries here only for custom domains, e.g. \"example.com\": \"216.239.38.120\".", + "hosts": {} +} diff --git a/domain_fronter.py b/domain_fronter.py new file mode 100644 index 0000000..a623cab --- /dev/null +++ b/domain_fronter.py @@ -0,0 +1,1230 @@ +""" +CDN Relay engine. + +Modes: + 1. custom_domain — SNI and Host both point to your custom domain on CF. + 2. domain_fronting — SNI = front_domain (allowed), Host = worker_host. + 3. google_fronting — Connect to Google IP, SNI=google, Host=Cloud Run. + 4. apps_script — Domain fronting via Google Apps Script relay. + POST JSON to script.google.com (fronted through www.google.com). + Apps Script fetches the target URL and returns the response. + +Modes 1-3: + tunnel() — WebSocket-based TCP tunnel (HTTPS / any TCP) + forward() — HTTP request forwarding (plain HTTP) + +Mode 4 (apps_script): + relay() — JSON-based HTTP relay through Apps Script +""" + +import asyncio +import base64 +import gzip +import json +import logging +import os +import re +import ssl +import time +from urllib.parse import urlparse + +from ws import ws_encode, ws_decode + +log = logging.getLogger("Fronter") + + +class DomainFronter: + def __init__(self, config: dict): + mode = config.get("mode", "domain_fronting") + + if mode == "custom_domain": + domain = config["custom_domain"] + self.connect_host = domain + self.sni_host = domain + self.http_host = domain + elif mode == "google_fronting": + self.connect_host = config.get("google_ip", "216.239.38.120") + self.sni_host = config.get("front_domain", "www.google.com") + self.http_host = config["worker_host"] + elif mode == "apps_script": + self.connect_host = config.get("google_ip", "216.239.38.120") + self.sni_host = config.get("front_domain", "www.google.com") + self.http_host = "script.google.com" + # Multi-script round-robin for higher throughput + script = config.get("script_ids") or config.get("script_id") + self._script_ids = script if isinstance(script, list) else [script] + self._script_idx = 0 + self.script_id = self._script_ids[0] # backward compat / logging + self._dev_available = False # True if /dev endpoint works (no redirect, ~400ms faster) + else: + self.connect_host = config["front_domain"] + self.sni_host = config["front_domain"] + self.http_host = config["worker_host"] + + self.mode = mode + self.worker_path = config.get("worker_path", "") + self.auth_key = config.get("auth_key", "") + self.verify_ssl = config.get("verify_ssl", True) + + # Connection pool — TTL-based, pre-warmed, with concurrency control + self._pool: list[tuple[asyncio.StreamReader, asyncio.StreamWriter, float]] = [] + self._pool_lock = asyncio.Lock() + self._pool_max = 50 + self._conn_ttl = 45.0 # seconds before a pooled conn is discarded + self._semaphore = asyncio.Semaphore(50) # max concurrent relay connections + self._warmed = False + self._refilling = False # background pool refill in progress + self._pool_min_idle = 15 # maintain at least this many idle connections + self._maintenance_task: asyncio.Task | None = None + + # Batch collector for grouping concurrent relay() calls + self._batch_lock = asyncio.Lock() + self._batch_pending: list[tuple[dict, asyncio.Future]] = [] + self._batch_task: asyncio.Task | None = None + self._batch_window_micro = 0.005 # 5ms micro-window (single req) + self._batch_window_macro = 0.050 # 50ms macro-window (burst traffic) + self._batch_max = 50 # max requests per batch + self._batch_enabled = True # disabled on first batch API failure + + # Request coalescing — dedup concurrent identical GETs + self._coalesce: dict[str, list[asyncio.Future]] = {} + + # HTTP/2 multiplexing — one connection handles all requests + self._h2 = None + if mode == "apps_script": + try: + from h2_transport import H2Transport, H2_AVAILABLE + if H2_AVAILABLE: + self._h2 = H2Transport( + self.connect_host, self.sni_host, self.verify_ssl + ) + log.info("HTTP/2 multiplexing available — " + "all requests will share one connection") + except ImportError: + pass + + # ── helpers ─────────────────────────────────────────────────── + + def _ssl_ctx(self) -> ssl.SSLContext: + ctx = ssl.create_default_context() + if not self.verify_ssl: + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return ctx + + async def _open(self): + """Open a TLS connection to the CDN. + + The *server_hostname* parameter sets the **TLS SNI** extension. + DPI systems see only this value. + """ + return await asyncio.open_connection( + self.connect_host, + 443, + ssl=self._ssl_ctx(), + server_hostname=self.sni_host, + ) + + async def _acquire(self): + """Get a healthy TLS connection from pool (TTL-checked) or open new.""" + now = asyncio.get_event_loop().time() + async with self._pool_lock: + while self._pool: + reader, writer, created = self._pool.pop() + if (now - created) < self._conn_ttl and not reader.at_eof(): + # Eagerly replace the connection we just took + asyncio.create_task(self._add_conn_to_pool()) + return reader, writer, created + try: + writer.close() + except Exception: + pass + reader, writer = await asyncio.wait_for(self._open(), timeout=10) + # Pool was empty — trigger aggressive background refill + if not self._refilling: + self._refilling = True + asyncio.create_task(self._refill_pool()) + return reader, writer, asyncio.get_event_loop().time() + + async def _release(self, reader, writer, created): + """Return a connection to the pool if still young and healthy.""" + now = asyncio.get_event_loop().time() + if (now - created) >= self._conn_ttl or reader.at_eof(): + try: + writer.close() + except Exception: + pass + return + async with self._pool_lock: + if len(self._pool) < self._pool_max: + self._pool.append((reader, writer, created)) + else: + try: + writer.close() + except Exception: + pass + + def _next_script_id(self) -> str: + """Round-robin across script IDs for load distribution.""" + sid = self._script_ids[self._script_idx % len(self._script_ids)] + self._script_idx += 1 + return sid + + def _exec_path(self) -> str: + """Get the next Apps Script endpoint path (/dev or /exec).""" + sid = self._next_script_id() + return f"/macros/s/{sid}/{'dev' if self._dev_available else 'exec'}" + + async def _flush_pool(self): + """Close all pooled connections (they may be stale after errors).""" + async with self._pool_lock: + for _, writer, _ in self._pool: + try: + writer.close() + except Exception: + pass + self._pool.clear() + + async def _refill_pool(self): + """Background: open connections in parallel to refill empty pool.""" + try: + coros = [self._add_conn_to_pool() for _ in range(8)] + await asyncio.gather(*coros, return_exceptions=True) + finally: + self._refilling = False + + async def _add_conn_to_pool(self): + """Open one TLS connection and add it to the pool.""" + try: + r, w = await asyncio.wait_for(self._open(), timeout=5) + t = asyncio.get_event_loop().time() + async with self._pool_lock: + if len(self._pool) < self._pool_max: + self._pool.append((r, w, t)) + else: + try: + w.close() + except Exception: + pass + except Exception: + pass + + async def _pool_maintenance(self): + """Continuously maintain healthy pool levels in background.""" + while True: + try: + await asyncio.sleep(3) + now = asyncio.get_event_loop().time() + + # Purge expired / dead connections + async with self._pool_lock: + alive = [] + for r, w, t in self._pool: + if (now - t) < self._conn_ttl and not r.at_eof(): + alive.append((r, w, t)) + else: + try: + w.close() + except Exception: + pass + self._pool = alive + idle = len(self._pool) + + # Refill if below minimum idle threshold + needed = max(0, self._pool_min_idle - idle) + if needed > 0: + coros = [self._add_conn_to_pool() + for _ in range(min(needed, 5))] + await asyncio.gather(*coros, return_exceptions=True) + + except asyncio.CancelledError: + break + except Exception: + pass + + async def _warm_pool(self): + """Pre-open TLS connections in the background. Never blocks relay().""" + if self._warmed: + return + self._warmed = True + asyncio.create_task(self._do_warm()) + # Start continuous pool maintenance + if self._maintenance_task is None: + self._maintenance_task = asyncio.create_task(self._pool_maintenance()) + # Start H2 connection (runs alongside H1 pool) + if self._h2: + asyncio.create_task(self._h2_connect_and_warm()) + + async def _h2_connect(self): + """Connect the HTTP/2 transport in background.""" + try: + await self._h2.ensure_connected() + log.info("H2 multiplexing active — one conn handles all requests") + except Exception as e: + log.warning("H2 connect failed (%s), using H1 pool fallback", e) + + async def _h2_connect_and_warm(self): + """Connect H2, pre-warm the Apps Script container, start keepalive.""" + await self._h2_connect() + if self._h2 and self._h2.is_connected: + asyncio.create_task(self._prewarm_script()) + asyncio.create_task(self._keepalive_loop()) + + async def _prewarm_script(self): + """Pre-warm Apps Script and detect /dev fast path (no redirect).""" + payload = json.dumps( + {"m": "HEAD", "u": "http://example.com/", "k": self.auth_key} + ).encode() + hdrs = {"content-type": "application/json"} + sid = self._script_ids[0] + + # Test /dev endpoint — returns data inline (no 302 redirect). + # If it works, saves ~400ms per request by eliminating one round trip. + try: + dev_path = f"/macros/s/{sid}/dev" + t0 = time.perf_counter() + status, _, body = await asyncio.wait_for( + self._h2.request( + method="POST", path=dev_path, host=self.http_host, + headers=hdrs, body=payload, + ), + timeout=15, + ) + dt = (time.perf_counter() - t0) * 1000 + data = json.loads(body.decode(errors="replace")) + if "s" in data: + self._dev_available = True + log.info("/dev fast path active (%.0fms, no redirect)", dt) + return + except Exception as e: + log.debug("/dev test failed: %s", e) + + # Fallback: warm up with /exec + try: + exec_path = f"/macros/s/{sid}/exec" + t0 = time.perf_counter() + await asyncio.wait_for( + self._h2.request( + method="POST", path=exec_path, host=self.http_host, + headers=hdrs, body=payload, + ), + timeout=15, + ) + dt = (time.perf_counter() - t0) * 1000 + log.info("Apps Script pre-warmed in %.0fms", dt) + except Exception as e: + log.debug("Pre-warm failed: %s", e) + + async def _keepalive_loop(self): + """Send periodic pings to keep Apps Script warm + H2 connection alive.""" + while True: + try: + await asyncio.sleep(240) # 4 minutes — saves ~90 quota hits/day vs 180s + # Google's container timeout is ~5 min idle + if not self._h2 or not self._h2.is_connected: + try: + await self._h2.reconnect() + except Exception: + continue + + # H2 PING to keep connection alive + await self._h2.ping() + + # Apps Script keepalive — warm the container + payload = {"m": "HEAD", "u": "http://example.com/", "k": self.auth_key} + path = self._exec_path() + t0 = time.perf_counter() + await asyncio.wait_for( + self._h2.request( + method="POST", path=path, host=self.http_host, + headers={"content-type": "application/json"}, + body=json.dumps(payload).encode(), + ), + timeout=20, + ) + dt = (time.perf_counter() - t0) * 1000 + log.debug("Keepalive ping: %.0fms", dt) + except asyncio.CancelledError: + break + except Exception as e: + log.debug("Keepalive failed: %s", e) + + async def _do_warm(self): + """Open connections in parallel — failures are fine.""" + count = 30 + coros = [self._add_conn_to_pool() for _ in range(count)] + results = await asyncio.gather(*coros, return_exceptions=True) + opened = sum(1 for r in results if not isinstance(r, Exception)) + log.info("Pre-warmed %d/%d TLS connections", opened, count) + + def _auth_header(self) -> str: + return f"X-Auth-Key: {self.auth_key}\r\n" if self.auth_key else "" + + # ── WebSocket tunnel (CONNECT / HTTPS) ──────────────────────── + + async def tunnel(self, target_host: str, target_port: int, + client_r: asyncio.StreamReader, + client_w: asyncio.StreamWriter): + """Tunnel raw TCP bytes through a domain-fronted WebSocket.""" + try: + remote_r, remote_w = await self._open() + except Exception as e: + log.error("TLS connect to %s failed: %s", self.connect_host, e) + return + + try: + # ---- WebSocket upgrade ---- + ws_key = base64.b64encode(os.urandom(16)).decode() + path = f"{self.worker_path}/tunnel?host={target_host}&port={target_port}" + handshake = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {self.http_host}\r\n" + f"Upgrade: websocket\r\n" + f"Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {ws_key}\r\n" + f"Sec-WebSocket-Version: 13\r\n" + f"{self._auth_header()}" + f"\r\n" + ) + remote_w.write(handshake.encode()) + await remote_w.drain() + + # Read the 101 Switching Protocols response + resp = b"" + while b"\r\n\r\n" not in resp: + chunk = await asyncio.wait_for(remote_r.read(4096), timeout=15) + if not chunk: + raise ConnectionError("No WebSocket handshake response") + resp += chunk + + status_line = resp.split(b"\r\n")[0] + if b"101" not in status_line: + raise ConnectionError( + f"WebSocket upgrade rejected: {status_line.decode(errors='replace')}" + ) + + log.info("Tunnel ready → %s:%d", target_host, target_port) + + # ---- bidirectional relay ---- + await asyncio.gather( + self._client_to_ws(client_r, remote_w), + self._ws_to_client(remote_r, client_w), + ) + + except Exception as e: + log.error("Tunnel error (%s:%d): %s", target_host, target_port, e) + finally: + try: + remote_w.close() + except Exception: + pass + + async def _client_to_ws(self, src: asyncio.StreamReader, + dst: asyncio.StreamWriter): + """Read plaintext from the browser, wrap in WS frames, send to CDN.""" + try: + while True: + data = await src.read(16384) + if not data: + # Send a WS close frame + dst.write(ws_encode(b"", opcode=0x08)) + await dst.drain() + break + dst.write(ws_encode(data)) + await dst.drain() + except (ConnectionError, asyncio.CancelledError): + pass + + async def _ws_to_client(self, src: asyncio.StreamReader, + dst: asyncio.StreamWriter): + """Read WS frames from CDN, unwrap, write plaintext to browser.""" + buf = b"" + try: + while True: + chunk = await src.read(16384) + if not chunk: + break + buf += chunk + while buf: + result = ws_decode(buf) + if result is None: + break # need more data + opcode, payload, consumed = result + buf = buf[consumed:] + if opcode == 0x08: # close + return + if payload: + dst.write(payload) + await dst.drain() + except (ConnectionError, asyncio.CancelledError): + pass + + # ── HTTP forwarding ─────────────────────────────────────────── + + async def forward(self, raw_request: bytes) -> bytes: + """Forward a plain HTTP request through the domain-fronted channel. + + Uses keep-alive connections from the pool for efficiency. + """ + try: + reader, writer, created = await self._acquire() + + # Wrap the original HTTP request inside a POST to the worker. + request = ( + f"POST {self.worker_path}/forward HTTP/1.1\r\n" + f"Host: {self.http_host}\r\n" + f"Content-Type: application/octet-stream\r\n" + f"Content-Length: {len(raw_request)}\r\n" + f"Connection: keep-alive\r\n" + f"{self._auth_header()}" + f"\r\n" + ) + writer.write(request.encode() + raw_request) + await writer.drain() + + status, resp_headers, resp_body = await self._read_http_response(reader) + + await self._release(reader, writer, created) + + # The worker wraps the target's response in its own HTTP + # envelope. The body IS the raw HTTP response from the target. + return resp_body + + except Exception as e: + log.error("Forward failed: %s", e) + return b"HTTP/1.1 502 Bad Gateway\r\n\r\nDomain fronting request failed\r\n" + + # ── Apps Script relay (apps_script mode) ────────────────────── + + async def relay(self, method: str, url: str, + headers: dict, body: bytes = b"") -> bytes: + """Relay an HTTP request through Apps Script. + + Features: + - Pre-warms TLS connections on first call + - Coalesces concurrent identical GET requests + - Batches concurrent calls via fetchAll() (40ms window) + - Retries once on connection failure + - Concurrency-limited via semaphore + + Returns a raw HTTP response (status + headers + body). + """ + if not self._warmed: + await self._warm_pool() + + payload = self._build_payload(method, url, headers, body) + + # Coalesce concurrent GETs for the same URL. + # CRITICAL: do NOT coalesce when a Range header is present — + # parallel range downloads MUST each hit the server independently. + has_range = False + if headers: + for k in headers: + if k.lower() == "range": + has_range = True + break + if method == "GET" and not body and not has_range: + return await self._coalesced_submit(url, payload) + + return await self._batch_submit(payload) + + async def _coalesced_submit(self, url: str, payload: dict) -> bytes: + """Dedup concurrent requests for the same URL (no Range header).""" + if url in self._coalesce: + # Another task is already fetching this URL — wait for it + future = asyncio.get_event_loop().create_future() + self._coalesce[url].append(future) + log.debug("Coalesced request: %s", url[:60]) + return await future + + self._coalesce[url] = [] + try: + result = await self._batch_submit(payload) + # Resolve all waiters + for f in self._coalesce.get(url, []): + if not f.done(): + f.set_result(result) + return result + except Exception as e: + for f in self._coalesce.get(url, []): + if not f.done(): + f.set_exception(e) + raise + finally: + self._coalesce.pop(url, None) + + async def relay_parallel(self, method: str, url: str, + headers: dict, body: bytes = b"", + chunk_size: int = 256 * 1024, + max_parallel: int = 16) -> bytes: + """Relay with parallel range acceleration for large downloads. + + Strategy: + 1. Send initial GET with Range: bytes=0-{message}
" + return ( + f"HTTP/1.1 {status} Error\r\n" + f"Content-Type: text/html\r\n" + f"Content-Length: {len(body)}\r\n" + f"\r\n" + f"{body}" + ).encode() diff --git a/h2_transport.py b/h2_transport.py new file mode 100644 index 0000000..3557840 --- /dev/null +++ b/h2_transport.py @@ -0,0 +1,419 @@ +""" +HTTP/2 multiplexed transport for domain-fronted connections. + +One TLS connection → many concurrent HTTP/2 streams → massive throughput. +Eliminates per-request TLS handshake overhead entirely. + +Instead of a pool of 30 HTTP/1.1 connections (each handling 1 request), +this uses a SINGLE HTTP/2 connection handling 100+ concurrent requests. + +Performance comparison: + HTTP/1.1 pool: 30 connections × 1 request = 30 concurrent requests max + HTTP/2 mux: 1 connection × 100 streams = 100 concurrent requests + +Requires: pip install h2 +""" + +import asyncio +import gzip +import logging +import socket +import ssl +from urllib.parse import urlparse + +log = logging.getLogger("H2") + +try: + import h2.connection + import h2.config + import h2.events + import h2.settings + H2_AVAILABLE = True +except ImportError: + H2_AVAILABLE = False + + +class _StreamState: + """State for a single in-flight HTTP/2 stream.""" + __slots__ = ("status", "headers", "data", "done", "error") + + def __init__(self): + self.status = 0 + self.headers: dict[str, str] = {} + self.data = bytearray() + self.done = asyncio.Event() + self.error: str | None = None + + +class H2Transport: + """ + Persistent HTTP/2 connection with automatic stream multiplexing. + + All relay requests share ONE TLS connection. Each request becomes + an independent HTTP/2 stream, running fully concurrently. + + Features: + - Auto-connect on first use + - Auto-reconnect on connection loss + - Redirect following (as new streams, same connection) + - Gzip decompression + - Configurable max concurrency + """ + + def __init__(self, connect_host: str, sni_host: str, + verify_ssl: bool = True): + self.connect_host = connect_host + self.sni_host = sni_host + self.verify_ssl = verify_ssl + + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._h2: "h2.connection.H2Connection | None" = None + self._connected = False + + self._write_lock = asyncio.Lock() + self._connect_lock = asyncio.Lock() + self._read_task: asyncio.Task | None = None + + # Per-stream tracking + self._streams: dict[int, _StreamState] = {} + + # Stats + self.total_requests = 0 + self.total_streams = 0 + + # ── Connection lifecycle ────────────────────────────────────── + + @property + def is_connected(self) -> bool: + return self._connected + + async def ensure_connected(self): + """Connect if not already connected.""" + if self._connected: + return + async with self._connect_lock: + if self._connected: + return + await self._do_connect() + + async def _do_connect(self): + """Establish the HTTP/2 connection with optimized socket settings.""" + ctx = ssl.create_default_context() + # Advertise both h2 and http/1.1 — some DPI blocks h2-only ALPN + ctx.set_alpn_protocols(["h2", "http/1.1"]) + if not self.verify_ssl: + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + + # Create raw TCP socket with TCP_NODELAY BEFORE TLS handshake. + # Nagle's algorithm can delay small writes (H2 frames) by up to 200ms + # waiting to coalesce — TCP_NODELAY forces immediate send. + raw = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + raw.setblocking(False) + + try: + await asyncio.wait_for( + asyncio.get_event_loop().sock_connect( + raw, (self.connect_host, 443) + ), + timeout=15, + ) + self._reader, self._writer = await asyncio.wait_for( + asyncio.open_connection( + ssl=ctx, + server_hostname=self.sni_host, + sock=raw, + ), + timeout=15, + ) + except Exception: + raw.close() + raise + + # Verify we actually got HTTP/2 + ssl_obj = self._writer.get_extra_info("ssl_object") + negotiated = ssl_obj.selected_alpn_protocol() if ssl_obj else None + if negotiated != "h2": + self._writer.close() + raise RuntimeError( + f"H2 ALPN negotiation failed (got {negotiated!r})" + ) + + config = h2.config.H2Configuration( + client_side=True, + header_encoding="utf-8", + ) + self._h2 = h2.connection.H2Connection(config=config) + self._h2.initiate_connection() + + # Connection-level flow control: ~16MB window + self._h2.increment_flow_control_window(2 ** 24 - 65535) + + # Per-stream settings: 1MB initial window, disable server push + self._h2.update_settings({ + h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: 1 * 1024 * 1024, + h2.settings.SettingCodes.ENABLE_PUSH: 0, + }) + + await self._flush() + + self._connected = True + self._read_task = asyncio.create_task(self._reader_loop()) + log.info("H2 connected → %s (SNI=%s, TCP_NODELAY=on)", + self.connect_host, self.sni_host) + + async def reconnect(self): + """Close current connection and re-establish.""" + await self._close_internal() + await self._do_connect() + + async def _close_internal(self): + self._connected = False + if self._read_task: + self._read_task.cancel() + self._read_task = None + if self._writer: + try: + self._writer.close() + except Exception: + pass + self._writer = None + # Wake all pending streams so they can raise + for state in self._streams.values(): + state.error = "Connection closed" + state.done.set() + self._streams.clear() + + # ── Public API ──────────────────────────────────────────────── + + async def request(self, method: str, path: str, host: str, + headers: dict | None = None, + body: bytes | None = None, + timeout: float = 25, + follow_redirects: int = 5) -> tuple[int, dict, bytes]: + """ + Send an HTTP/2 request and return (status, headers, body). + + Thread-safe: many concurrent calls each get their own stream. + Redirects are followed as new streams on the same connection. + """ + await self.ensure_connected() + self.total_requests += 1 + + for _ in range(follow_redirects + 1): + status, resp_headers, resp_body = await self._single_request( + method, path, host, headers, body, timeout, + ) + + if status not in (301, 302, 303, 307, 308): + return status, resp_headers, resp_body + + location = resp_headers.get("location", "") + if not location: + return status, resp_headers, resp_body + + parsed = urlparse(location) + path = parsed.path + ("?" + parsed.query if parsed.query else "") + host = parsed.netloc or host + method = "GET" + body = None + headers = None # Drop request headers on redirect + + return status, resp_headers, resp_body + + # ── Stream handling ─────────────────────────────────────────── + + async def _single_request(self, method, path, host, headers, body, + timeout) -> tuple[int, dict, bytes]: + """Send one HTTP/2 request on a new stream, wait for response.""" + if not self._connected: + await self.ensure_connected() + + stream_id = None + + async with self._write_lock: + try: + stream_id = self._h2.get_next_available_stream_id() + except Exception: + # Connection is stale — reconnect + await self.reconnect() + stream_id = self._h2.get_next_available_stream_id() + + h2_headers = [ + (":method", method), + (":path", path), + (":authority", host), + (":scheme", "https"), + ("accept-encoding", "gzip"), + ] + if headers: + for k, v in headers.items(): + h2_headers.append((k.lower(), str(v))) + + end_stream = not body + self._h2.send_headers(stream_id, h2_headers, end_stream=end_stream) + + if body: + # Send body (may need chunking for flow control) + self._send_body(stream_id, body) + + state = _StreamState() + self._streams[stream_id] = state + self.total_streams += 1 + + await self._flush() + + # Wait for complete response + try: + await asyncio.wait_for(state.done.wait(), timeout=timeout) + except asyncio.TimeoutError: + self._streams.pop(stream_id, None) + raise TimeoutError( + f"H2 stream {stream_id} timed out ({timeout}s)" + ) + + self._streams.pop(stream_id, None) + + if state.error: + raise ConnectionError(f"H2 stream error: {state.error}") + + # Auto-decompress gzip + resp_body = bytes(state.data) + if state.headers.get("content-encoding", "").lower() == "gzip": + try: + resp_body = gzip.decompress(resp_body) + except Exception: + pass + + return state.status, state.headers, resp_body + + def _send_body(self, stream_id: int, body: bytes): + """Send request body, respecting H2 flow control window.""" + # For small bodies (typical JSON payloads), send in one shot + while body: + max_size = self._h2.local_settings.max_frame_size + window = self._h2.local_flow_control_window(stream_id) + send_size = min(len(body), max_size, window) + if send_size <= 0: + # Flow control full — let the reader loop process + # window updates before we continue + break + end = send_size >= len(body) + self._h2.send_data(stream_id, body[:send_size], end_stream=end) + body = body[send_size:] + + # ── Background reader ───────────────────────────────────────── + + async def _reader_loop(self): + """Background: read H2 frames, dispatch events to waiting streams.""" + try: + while self._connected: + data = await self._reader.read(65536) + if not data: + log.warning("H2 remote closed connection") + break + + try: + events = self._h2.receive_data(data) + except Exception as e: + log.error("H2 protocol error: %s", e) + break + + for event in events: + self._dispatch(event) + + # Send pending data (acks, window updates, ping responses) + async with self._write_lock: + await self._flush() + + except asyncio.CancelledError: + pass + except Exception as e: + log.error("H2 reader error: %s", e) + finally: + self._connected = False + for state in self._streams.values(): + if not state.done.is_set(): + state.error = "Connection lost" + state.done.set() + log.info("H2 reader loop ended") + + def _dispatch(self, event): + """Route a single h2 event to its stream.""" + if isinstance(event, h2.events.ResponseReceived): + state = self._streams.get(event.stream_id) + if state: + for name, value in event.headers: + n = name if isinstance(name, str) else name.decode() + v = value if isinstance(value, str) else value.decode() + if n == ":status": + state.status = int(v) + else: + state.headers[n] = v + + elif isinstance(event, h2.events.DataReceived): + state = self._streams.get(event.stream_id) + if state: + state.data.extend(event.data) + # Always acknowledge received data for flow control + self._h2.acknowledge_received_data( + event.flow_controlled_length, event.stream_id + ) + + elif isinstance(event, h2.events.StreamEnded): + state = self._streams.get(event.stream_id) + if state: + state.done.set() + + elif isinstance(event, h2.events.StreamReset): + state = self._streams.get(event.stream_id) + if state: + state.error = f"Stream reset (code={event.error_code})" + state.done.set() + + elif isinstance(event, h2.events.WindowUpdated): + pass # h2 library handles window bookkeeping + + elif isinstance(event, h2.events.SettingsAcknowledged): + pass + + elif isinstance(event, h2.events.PingReceived): + pass # h2 library auto-responds + + elif isinstance(event, h2.events.PingAckReceived): + pass # keepalive confirmed + + # ── Internal ────────────────────────────────────────────────── + + async def _flush(self): + """Write pending H2 frame data to the socket.""" + data = self._h2.data_to_send() + if data and self._writer: + self._writer.write(data) + await self._writer.drain() + + async def close(self): + """Gracefully close the HTTP/2 connection.""" + if self._h2 and self._connected: + try: + self._h2.close_connection() + async with self._write_lock: + await self._flush() + except Exception: + pass + await self._close_internal() + + async def ping(self): + """Send an H2 PING frame to keep the connection alive.""" + if not self._connected or not self._h2: + return + try: + async with self._write_lock: + if not self._connected: + return + self._h2.ping(b"\x00" * 8) + await self._flush() + except Exception as e: + log.debug("H2 PING failed: %s", e) diff --git a/main.py b/main.py new file mode 100644 index 0000000..d80add6 --- /dev/null +++ b/main.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +""" +DomainFront Tunnel — Bypass DPI censorship via Domain Fronting. + +Run a local HTTP proxy that tunnels all traffic through a CDN using +domain fronting: the TLS SNI shows an allowed domain while the encrypted +HTTP Host header routes to your Cloudflare Worker relay. +""" + +import argparse +import asyncio +import json +import logging +import os +import sys + +from cert_installer import install_ca, is_ca_trusted +from mitm import CA_CERT_FILE +from proxy_server import ProxyServer + +__version__ = "1.0.0" + + +def setup_logging(level_name: str): + level = getattr(logging, level_name.upper(), logging.INFO) + logging.basicConfig( + level=level, + format="%(asctime)s [%(name)-12s] %(levelname)-7s %(message)s", + datefmt="%H:%M:%S", + ) + + +def parse_args(): + parser = argparse.ArgumentParser( + prog="domainfront-tunnel", + description="Local HTTP proxy that tunnels traffic through domain fronting.", + ) + parser.add_argument( + "-c", "--config", + default=os.environ.get("DFT_CONFIG", "config.json"), + help="Path to config file (default: config.json, env: DFT_CONFIG)", + ) + parser.add_argument( + "-p", "--port", + type=int, + default=None, + help="Override listen port (env: DFT_PORT)", + ) + parser.add_argument( + "--host", + default=None, + help="Override listen host (env: DFT_HOST)", + ) + parser.add_argument( + "--log-level", + choices=["DEBUG", "INFO", "WARNING", "ERROR"], + default=None, + help="Override log level (env: DFT_LOG_LEVEL)", + ) + parser.add_argument( + "-v", "--version", + action="version", + version=f"%(prog)s {__version__}", + ) + parser.add_argument( + "--install-cert", + action="store_true", + help="Install the MITM CA certificate as a trusted root and exit.", + ) + parser.add_argument( + "--no-cert-check", + action="store_true", + help="Skip the certificate installation check on startup.", + ) + return parser.parse_args() + + +def main(): + args = parse_args() + config_path = args.config + + try: + with open(config_path) as f: + config = json.load(f) + except FileNotFoundError: + print(f"Config not found: {config_path}") + print("Copy config.example.json to config.json and fill in your values.") + sys.exit(1) + except json.JSONDecodeError as e: + print(f"Invalid JSON in config: {e}") + sys.exit(1) + + # Environment variable overrides + if os.environ.get("DFT_AUTH_KEY"): + config["auth_key"] = os.environ["DFT_AUTH_KEY"] + if os.environ.get("DFT_SCRIPT_ID"): + config["script_id"] = os.environ["DFT_SCRIPT_ID"] + + # CLI argument overrides + if args.port is not None: + config["listen_port"] = args.port + elif os.environ.get("DFT_PORT"): + config["listen_port"] = int(os.environ["DFT_PORT"]) + + if args.host is not None: + config["listen_host"] = args.host + elif os.environ.get("DFT_HOST"): + config["listen_host"] = os.environ["DFT_HOST"] + + if args.log_level is not None: + config["log_level"] = args.log_level + elif os.environ.get("DFT_LOG_LEVEL"): + config["log_level"] = os.environ["DFT_LOG_LEVEL"] + + for key in ("auth_key",): + if key not in config: + print(f"Missing required config key: {key}") + sys.exit(1) + + mode = config.get("mode", "domain_fronting") + if mode == "custom_domain" and "custom_domain" not in config: + print("Mode 'custom_domain' requires 'custom_domain' in config") + sys.exit(1) + if mode == "domain_fronting": + for key in ("front_domain", "worker_host"): + if key not in config: + print(f"Mode 'domain_fronting' requires '{key}' in config") + sys.exit(1) + if mode == "google_fronting": + if "worker_host" not in config: + print("Mode 'google_fronting' requires 'worker_host' in config (your Cloud Run URL)") + sys.exit(1) + if mode == "apps_script": + sid = config.get("script_ids") or config.get("script_id") + if not sid or (isinstance(sid, str) and sid == "YOUR_APPS_SCRIPT_DEPLOYMENT_ID"): + print("Mode 'apps_script' requires 'script_id' in config.") + print("Deploy the Apps Script from appsscript/Code.gs and paste the Deployment ID.") + sys.exit(1) + + # ── Certificate installation ────────────────────────────────────────── + if args.install_cert: + setup_logging("INFO") + _log = logging.getLogger("Main") + _log.info("Installing CA certificate…") + ok = install_ca(CA_CERT_FILE) + sys.exit(0 if ok else 1) + + setup_logging(config.get("log_level", "INFO")) + log = logging.getLogger("Main") + + mode = config.get("mode", "domain_fronting") + log.info("DomainFront Tunnel starting (mode: %s)", mode) + + if mode == "custom_domain": + log.info("Custom domain : %s", config["custom_domain"]) + elif mode == "google_fronting": + log.info("Google fronting : SNI=%s → Host=%s", + config.get("front_domain", "www.google.com"), config["worker_host"]) + log.info("Google IP : %s", config.get("google_ip", "216.239.38.120")) + elif mode == "apps_script": + log.info("Apps Script relay : SNI=%s → script.google.com", + config.get("front_domain", "www.google.com")) + script_ids = config.get("script_ids") or config.get("script_id") + if isinstance(script_ids, list): + log.info("Script IDs : %d scripts (round-robin)", len(script_ids)) + for i, sid in enumerate(script_ids): + log.info(" [%d] %s", i + 1, sid) + else: + log.info("Script ID : %s", script_ids) + + # Ensure CA file exists before checking / installing it. + # MITMCertManager generates ca/ca.crt on first instantiation. + if not os.path.exists(CA_CERT_FILE): + from mitm import MITMCertManager + MITMCertManager() # side-effect: creates ca/ca.crt + ca/ca.key + + # Auto-install MITM CA if not already trusted + if not args.no_cert_check: + if not is_ca_trusted(CA_CERT_FILE): + log.warning("MITM CA is not trusted — attempting automatic installation…") + ok = install_ca(CA_CERT_FILE) + if ok: + log.info("CA certificate installed. You may need to restart your browser.") + else: + log.error( + "Auto-install failed. Run with --install-cert (may need admin/sudo) " + "or manually install ca/ca.crt as a trusted root CA." + ) + else: + log.info("MITM CA is already trusted.") + else: + log.info("Front domain (SNI) : %s", config.get("front_domain", "?")) + log.info("Worker host (Host) : %s", config.get("worker_host", "?")) + + log.info("Proxy address : %s:%d", config.get("listen_host", "127.0.0.1"), config.get("listen_port", 8080)) + + try: + asyncio.run(ProxyServer(config).start()) + except KeyboardInterrupt: + log.info("Stopped") + + +if __name__ == "__main__": + main() diff --git a/mitm.py b/mitm.py new file mode 100644 index 0000000..e3c91b1 --- /dev/null +++ b/mitm.py @@ -0,0 +1,153 @@ +""" +MITM certificate manager for HTTPS interception. + +Generates a CA certificate (once, stored as files) and per-domain +certificates (on the fly, cached in memory) so the local proxy can +decrypt HTTPS traffic and relay it through Apps Script. + +The user must install ca/ca.crt in their browser's trusted CAs once. + +Requires: pip install cryptography +""" + +import datetime +import logging +import os +import ssl +import tempfile + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + +log = logging.getLogger("MITM") + +CA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ca") +CA_KEY_FILE = os.path.join(CA_DIR, "ca.key") +CA_CERT_FILE = os.path.join(CA_DIR, "ca.crt") + + +class MITMCertManager: + def __init__(self): + self._ca_key = None + self._ca_cert = None + self._ctx_cache: dict[str, ssl.SSLContext] = {} + self._cert_dir = tempfile.mkdtemp(prefix="domainfront_certs_") + self._ensure_ca() + + def _ensure_ca(self): + if os.path.exists(CA_KEY_FILE) and os.path.exists(CA_CERT_FILE): + with open(CA_KEY_FILE, "rb") as f: + self._ca_key = serialization.load_pem_private_key( + f.read(), password=None + ) + with open(CA_CERT_FILE, "rb") as f: + self._ca_cert = x509.load_pem_x509_certificate(f.read()) + log.info("Loaded CA from %s", CA_DIR) + else: + self._create_ca() + + def _create_ca(self): + os.makedirs(CA_DIR, exist_ok=True) + + self._ca_key = rsa.generate_private_key( + public_exponent=65537, key_size=2048 + ) + subject = issuer = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, "MasterHttpRelayVPN"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MasterHttpRelayVPN"), + ]) + now = datetime.datetime.now(datetime.timezone.utc) + self._ca_cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(self._ca_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now + datetime.timedelta(days=3650)) + .add_extension( + x509.BasicConstraints(ca=True, path_length=0), critical=True + ) + .add_extension( + x509.KeyUsage( + digital_signature=True, + key_cert_sign=True, + crl_sign=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(self._ca_key, hashes.SHA256()) + ) + + with open(CA_KEY_FILE, "wb") as f: + f.write( + self._ca_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + with open(CA_CERT_FILE, "wb") as f: + f.write(self._ca_cert.public_bytes(serialization.Encoding.PEM)) + + log.warning("Generated new CA certificate: %s", CA_CERT_FILE) + log.warning(">>> Install this file in your browser's Trusted Root CAs! <<<") + + def get_server_context(self, domain: str) -> ssl.SSLContext: + if domain not in self._ctx_cache: + key_pem, cert_pem = self._generate_domain_cert(domain) + + cert_file = os.path.join(self._cert_dir, f"{domain}.crt") + key_file = os.path.join(self._cert_dir, f"{domain}.key") + + ca_pem = self._ca_cert.public_bytes(serialization.Encoding.PEM) + with open(cert_file, "wb") as f: + f.write(cert_pem + ca_pem) + with open(key_file, "wb") as f: + f.write(key_pem) + + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.set_alpn_protocols(["http/1.1"]) + ctx.load_cert_chain(cert_file, key_file) + self._ctx_cache[domain] = ctx + + return self._ctx_cache[domain] + + def _generate_domain_cert(self, domain: str): + key = rsa.generate_private_key( + public_exponent=65537, key_size=2048 + ) + subject = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, domain), + ]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(self._ca_cert.subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now + datetime.timedelta(days=365)) + .add_extension( + x509.SubjectAlternativeName([x509.DNSName(domain)]), + critical=False, + ) + .sign(self._ca_key, hashes.SHA256()) + ) + + key_pem = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + cert_pem = cert.public_bytes(serialization.Encoding.PEM) + return key_pem, cert_pem diff --git a/proxy_server.py b/proxy_server.py new file mode 100644 index 0000000..fb9b1d8 --- /dev/null +++ b/proxy_server.py @@ -0,0 +1,777 @@ +""" +Local HTTP proxy server. + +Intercepts the user's browser traffic and forwards everything through +a domain-fronted connection to a CDN worker or Apps Script relay. + +Supports: + - CONNECT method → WebSocket tunnel (modes 1-3) or MITM relay (apps_script) + - GET / POST etc. → HTTP forwarding (modes 1-3) or JSON relay (apps_script) +""" + +import asyncio +import logging +import re +import ssl +import time + +from domain_fronter import DomainFronter + +log = logging.getLogger("Proxy") + + +class ResponseCache: + """Simple LRU response cache — avoids repeated relay calls.""" + + def __init__(self, max_mb: int = 50): + self._store: dict[str, tuple[bytes, float]] = {} + self._size = 0 + self._max = max_mb * 1024 * 1024 + self.hits = 0 + self.misses = 0 + + def get(self, url: str) -> bytes | None: + entry = self._store.get(url) + if not entry: + self.misses += 1 + return None + raw, expires = entry + if time.time() > expires: + self._size -= len(raw) + del self._store[url] + self.misses += 1 + return None + self.hits += 1 + return raw + + def put(self, url: str, raw_response: bytes, ttl: int = 300): + size = len(raw_response) + if size > self._max // 4 or size == 0: + return + # Evict oldest to make room + while self._size + size > self._max and self._store: + oldest = next(iter(self._store)) + self._size -= len(self._store[oldest][0]) + del self._store[oldest] + if url in self._store: + self._size -= len(self._store[url][0]) + self._store[url] = (raw_response, time.time() + ttl) + self._size += size + + @staticmethod + def parse_ttl(raw_response: bytes, url: str) -> int: + """Determine cache TTL from response headers and URL.""" + hdr_end = raw_response.find(b"\r\n\r\n") + if hdr_end < 0: + return 0 + hdr = raw_response[:hdr_end].decode(errors="replace").lower() + + # Don't cache errors or non-200 + if b"HTTP/1.1 200" not in raw_response[:20]: + return 0 + if "no-store" in hdr: + return 0 + + # Explicit max-age + m = re.search(r"max-age=(\d+)", hdr) + if m: + return min(int(m.group(1)), 86400) + + # Heuristic by content type / extension + path = url.split("?")[0].lower() + static_exts = ( + ".css", ".js", ".woff", ".woff2", ".ttf", ".eot", + ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".ico", + ".mp3", ".mp4", ".wasm", + ) + for ext in static_exts: + if path.endswith(ext): + return 3600 # 1 hour for static assets + + ct_m = re.search(r"content-type:\s*([^\r\n]+)", hdr) + ct = ct_m.group(1) if ct_m else "" + if "image/" in ct or "font/" in ct: + return 3600 + if "text/css" in ct or "javascript" in ct: + return 1800 + if "text/html" in ct or "application/json" in ct: + return 0 # don't cache dynamic content by default + + return 0 + + +class ProxyServer: + def __init__(self, config: dict): + self.host = config.get("listen_host", "127.0.0.1") + self.port = config.get("listen_port", 8080) + self.mode = config.get("mode", "domain_fronting") + self.fronter = DomainFronter(config) + self.mitm = None + self._cache = ResponseCache(max_mb=50) + + # Persistent HTTP tunnel cache for google_fronting mode + # Key: "host:port" → (tunnel_reader, tunnel_writer, lock) + self._http_tunnels: dict = {} + self._tunnel_lock = asyncio.Lock() + + # hosts override — DNS fake-map: domain/suffix → IP + # Checked before any real DNS lookup; supports exact and suffix matching. + self._hosts: dict[str, str] = config.get("hosts", {}) + + if self.mode == "apps_script": + try: + from mitm import MITMCertManager + self.mitm = MITMCertManager() + except ImportError: + log.error("apps_script mode requires 'cryptography' package.") + log.error("Run: pip install cryptography") + raise SystemExit(1) + + async def start(self): + srv = await asyncio.start_server(self._on_client, self.host, self.port) + log.info( + "Listening on %s:%d — configure your browser HTTP proxy to this address", + self.host, self.port, + ) + async with srv: + await srv.serve_forever() + + # ── client handler ──────────────────────────────────────────── + + async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + addr = writer.get_extra_info("peername") + try: + first_line = await asyncio.wait_for(reader.readline(), timeout=30) + if not first_line: + return + + # Read remaining headers + header_block = first_line + while True: + line = await asyncio.wait_for(reader.readline(), timeout=10) + header_block += line + if line in (b"\r\n", b"\n", b""): + break + + request_line = first_line.decode(errors="replace").strip() + parts = request_line.split(" ", 2) + if len(parts) < 2: + return + + method = parts[0].upper() + + if method == "CONNECT": + await self._do_connect(parts[1], reader, writer) + else: + await self._do_http(header_block, reader, writer) + + except asyncio.TimeoutError: + log.debug("Timeout: %s", addr) + except Exception as e: + log.error("Error (%s): %s", addr, e) + finally: + try: + writer.close() + await writer.wait_closed() + except Exception: + pass + + # ── CONNECT (HTTPS tunnelling) ──────────────────────────────── + + async def _do_connect(self, target: str, reader, writer): + host, _, port = target.rpartition(":") + port = int(port) if port else 443 + if not host: + host, port = target, 443 + + log.info("CONNECT → %s:%d", host, port) + + writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n") + await writer.drain() + + if self.mode == "apps_script": + override_ip = self._sni_rewrite_ip(host) + if override_ip: + # SNI-blocked domain: MITM-decrypt from browser, then + # re-connect to the override IP with SNI=front_domain so + # the ISP never sees the blocked hostname in the TLS handshake. + log.info("SNI-rewrite tunnel → %s via %s (SNI: %s)", + host, override_ip, self.fronter.sni_host) + await self._do_sni_rewrite_tunnel(host, port, reader, writer, + connect_ip=override_ip) + elif self._is_google_domain(host): + log.info("Direct tunnel → %s (Google domain, skipping relay)", host) + await self._do_direct_tunnel(host, port, reader, writer) + else: + await self._do_mitm_connect(host, port, reader, writer) + else: + await self.fronter.tunnel(host, port, reader, writer) + + # ── Hosts override (fake DNS) ───────────────────────────────── + + # Built-in list of domains that must be reached via Google's frontend IP + # with SNI rewritten to `front_domain` (default: www.google.com). + # These are Google-owned services whose real SNI is DPI-blocked in some + # countries, but that Google serves from the same edge IP as www.google.com. + # Users don't need to configure anything — any host matching one of these + # suffixes is transparently SNI-rewritten to the configured `google_ip`. + # Config's "hosts" map still takes precedence (for custom overrides). + _SNI_REWRITE_SUFFIXES = ( + "youtube.com", + "youtu.be", + "youtube-nocookie.com", + "ytimg.com", + "ggpht.com", + "gvt1.com", + "gvt2.com", + "doubleclick.net", + "googlesyndication.com", + "googleadservices.com", + "google-analytics.com", + "googletagmanager.com", + "googletagservices.com", + "fonts.googleapis.com", + ) + + def _sni_rewrite_ip(self, host: str) -> str | None: + """Return the IP to SNI-rewrite `host` through, or None. + + Order of precedence: + 1. Explicit entry in config `hosts` map (exact or suffix match). + 2. Built-in `_SNI_REWRITE_SUFFIXES` → mapped to config `google_ip`. + """ + ip = self._hosts_ip(host) + if ip: + return ip + h = host.lower().rstrip(".") + for suffix in self._SNI_REWRITE_SUFFIXES: + if h == suffix or h.endswith("." + suffix): + return self.fronter.connect_host # configured google_ip + return None + + def _hosts_ip(self, host: str) -> str | None: + """Return override IP for host if defined in config 'hosts', else None. + + Supports exact match and suffix match (e.g. 'youtube.com' matches + 'www.youtube.com', 'm.youtube.com', etc.). + """ + h = host.lower().rstrip(".") + if h in self._hosts: + return self._hosts[h] + # suffix match: check every parent label + parts = h.split(".") + for i in range(1, len(parts)): + parent = ".".join(parts[i:]) + if parent in self._hosts: + return self._hosts[parent] + return None + + # ── Google domain detection ─────────────────────────────────── + + # Only domains whose SNI the ISP does NOT block — direct tunnel is safe. + # YouTube/googlevideo SNIs are blocked; they go through _do_sni_rewrite_tunnel + # via the hosts map instead. + _GOOGLE_SUFFIXES = ( + ".google.com", ".google.co", + ".googleapis.com", ".gstatic.com", + ".googleusercontent.com", + ) + _GOOGLE_EXACT = { + "google.com", "gstatic.com", "googleapis.com", + } + + def _is_google_domain(self, host: str) -> bool: + """Return True if host is a Google-owned domain.""" + h = host.lower().rstrip(".") + if h in self._GOOGLE_EXACT: + return True + for suffix in self._GOOGLE_SUFFIXES: + if h.endswith(suffix): + return True + return False + + # ── Direct tunnel (no MITM) ─────────────────────────────────── + + async def _do_direct_tunnel(self, host: str, port: int, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + connect_ip: str | None = None): + """Pipe raw TLS bytes directly to the target server. + + connect_ip overrides DNS: the TCP connection goes to that IP + while the browser's TLS (SNI=host) is piped through unchanged. + Defaults to the configured google_ip for Google-category domains. + """ + target_ip = connect_ip or self.fronter.connect_host + try: + r_remote, w_remote = await asyncio.wait_for( + asyncio.open_connection(target_ip, port), timeout=10 + ) + except Exception as e: + log.error("Direct tunnel connect failed (%s via %s): %s", + host, target_ip, e) + return + + async def pipe(src, dst, label): + try: + while True: + data = await src.read(65536) + if not data: + break + dst.write(data) + await dst.drain() + except (ConnectionError, asyncio.CancelledError): + pass + except Exception as e: + log.debug("Pipe %s ended: %s", label, e) + finally: + try: + dst.close() + except Exception: + pass + + await asyncio.gather( + pipe(reader, w_remote, f"client→{host}"), + pipe(r_remote, writer, f"{host}→client"), + ) + + # ── SNI-rewrite tunnel ──────────────────────────────────────── + + async def _do_sni_rewrite_tunnel(self, host: str, port: int, reader, writer, + connect_ip: str | None = None): + """MITM-decrypt TLS from browser, then re-encrypt toward connect_ip + using SNI=front_domain (e.g. www.google.com). + + The ISP only ever sees SNI=www.google.com in the outgoing handshake, + hiding the blocked hostname (e.g. www.youtube.com). + """ + target_ip = connect_ip or self.fronter.connect_host + sni_out = self.fronter.sni_host # e.g. "www.google.com" + + # Step 1: MITM — accept TLS from the browser + ssl_ctx_server = self.mitm.get_server_context(host) + loop = asyncio.get_event_loop() + transport = writer.transport + protocol = transport.get_protocol() + try: + new_transport = await loop.start_tls( + transport, protocol, ssl_ctx_server, server_side=True, + ) + except Exception as e: + log.debug("SNI-rewrite TLS accept failed (%s): %s", host, e) + return + writer._transport = new_transport + + # Step 2: open outgoing TLS to target IP with the safe SNI + ssl_ctx_client = ssl.create_default_context() + if not self.fronter.verify_ssl: + ssl_ctx_client.check_hostname = False + ssl_ctx_client.verify_mode = ssl.CERT_NONE + try: + r_out, w_out = await asyncio.wait_for( + asyncio.open_connection( + target_ip, port, + ssl=ssl_ctx_client, + server_hostname=sni_out, + ), + timeout=10, + ) + except Exception as e: + log.error("SNI-rewrite outbound connect failed (%s via %s): %s", + host, target_ip, e) + return + + # Step 3: pipe application-layer bytes between the two TLS sessions + async def pipe(src, dst, label): + try: + while True: + data = await src.read(65536) + if not data: + break + dst.write(data) + await dst.drain() + except (ConnectionError, asyncio.CancelledError): + pass + except Exception as exc: + log.debug("Pipe %s ended: %s", label, exc) + finally: + try: + dst.close() + except Exception: + pass + + await asyncio.gather( + pipe(reader, w_out, f"client→{host}"), + pipe(r_out, writer, f"{host}→client"), + ) + + # ── MITM CONNECT (apps_script mode) ─────────────────────────── + + async def _do_mitm_connect(self, host: str, port: int, reader, writer): + """Intercept TLS, decrypt HTTP, and relay through Apps Script.""" + ssl_ctx = self.mitm.get_server_context(host) + + # Upgrade the existing connection to TLS (we are the server) + loop = asyncio.get_event_loop() + transport = writer.transport + protocol = transport.get_protocol() + + try: + new_transport = await loop.start_tls( + transport, protocol, ssl_ctx, server_side=True, + ) + except Exception as e: + # Non-HTTPS traffic (e.g. MTProto, plain HTTP on port 80/443) + # routed through the proxy will always fail TLS — log at DEBUG + # to avoid alarming noise. + if port != 443: + log.debug("TLS handshake skipped for %s:%d (non-HTTPS): %s", host, port, e) + else: + log.debug("TLS handshake failed for %s: %s", host, e) + return + + # Update writer to use the new TLS transport + writer._transport = new_transport + + # Read and relay HTTP requests from the browser (now decrypted) + while True: + try: + first_line = await asyncio.wait_for(reader.readline(), timeout=120) + if not first_line: + break + + header_block = first_line + while True: + line = await asyncio.wait_for(reader.readline(), timeout=10) + header_block += line + if line in (b"\r\n", b"\n", b""): + break + + # Read body + body = b"" + for raw_line in header_block.split(b"\r\n"): + if raw_line.lower().startswith(b"content-length:"): + length = int(raw_line.split(b":", 1)[1].strip()) + body = await reader.readexactly(length) + break + + # Parse the request + request_line = first_line.decode(errors="replace").strip() + parts = request_line.split(" ", 2) + if len(parts) < 2: + break + + method = parts[0] + path = parts[1] + + # Parse headers + headers = {} + for raw_line in header_block.split(b"\r\n")[1:]: + if b":" in raw_line: + k, v = raw_line.decode(errors="replace").split(":", 1) + headers[k.strip()] = v.strip() + + # Build full URL (browser sends just the path in CONNECT) + if port == 443: + url = f"https://{host}{path}" + else: + url = f"https://{host}:{port}{path}" + + log.info("MITM → %s %s", method, url) + + # ── CORS: extract relevant request headers ──────────────────── + origin = next( + (v for k, v in headers.items() if k.lower() == "origin"), "" + ) + acr_method = next( + (v for k, v in headers.items() + if k.lower() == "access-control-request-method"), "" + ) + acr_headers = next( + (v for k, v in headers.items() + if k.lower() == "access-control-request-headers"), "" + ) + + # CORS preflight — respond directly; UrlFetchApp doesn't + # support OPTIONS so forwarding it would always fail. + if method.upper() == "OPTIONS" and acr_method: + log.debug("CORS preflight → %s (responding locally)", url[:60]) + writer.write(self._cors_preflight_response(origin, acr_method, acr_headers)) + await writer.drain() + continue + + # Check local cache first (GET only) + response = None + if method == "GET" and not body: + response = self._cache.get(url) + if response: + log.debug("Cache HIT: %s", url[:60]) + + if response is None: + # Relay through Apps Script + try: + response = await self._relay_smart(method, url, headers, body) + except Exception as e: + log.error("Relay error (%s): %s", url[:60], e) + err_body = f"Relay error: {e}".encode() + response = ( + b"HTTP/1.1 502 Bad Gateway\r\n" + b"Content-Type: text/plain\r\n" + b"Content-Length: " + str(len(err_body)).encode() + b"\r\n" + b"\r\n" + err_body + ) + + # Cache successful GET responses + if method == "GET" and not body and response: + ttl = ResponseCache.parse_ttl(response, url) + if ttl > 0: + self._cache.put(url, response, ttl) + log.debug("Cached (%ds): %s", ttl, url[:60]) + + # Inject permissive CORS headers whenever the browser + # sent an Origin (cross-origin XHR / fetch). + if origin and response: + response = self._inject_cors_headers(response, origin) + + writer.write(response) + await writer.drain() + + except asyncio.TimeoutError: + break + except asyncio.IncompleteReadError: + break + except ConnectionError: + break + except Exception as e: + log.error("MITM handler error (%s): %s", host, e) + break + + # ── CORS helpers ────────────────────────────────────────────────────────── + + @staticmethod + def _cors_preflight_response(origin: str, acr_method: str, acr_headers: str) -> bytes: + """Return a 204 No Content response that satisfies a CORS preflight.""" + allow_origin = origin or "*" + allow_methods = ( + f"{acr_method}, GET, POST, PUT, DELETE, PATCH, OPTIONS" + if acr_method else + "GET, POST, PUT, DELETE, PATCH, OPTIONS" + ) + allow_headers = acr_headers or "*" + return ( + "HTTP/1.1 204 No Content\r\n" + f"Access-Control-Allow-Origin: {allow_origin}\r\n" + f"Access-Control-Allow-Methods: {allow_methods}\r\n" + f"Access-Control-Allow-Headers: {allow_headers}\r\n" + "Access-Control-Allow-Credentials: true\r\n" + "Access-Control-Max-Age: 86400\r\n" + "Vary: Origin\r\n" + "Content-Length: 0\r\n" + "\r\n" + ).encode() + + @staticmethod + def _inject_cors_headers(response: bytes, origin: str) -> bytes: + """Inject CORS headers only if the upstream response lacks them. + + We must NOT overwrite the origin server's CORS headers: sites like + x.com return carefully-scoped Access-Control-Allow-Headers that list + specific custom headers (e.g. x-csrf-token). Replacing them with + wildcards together with Allow-Credentials: true makes browsers + reject the response (per the Fetch spec, "*" is literal when + credentials are included), which the site then blames on privacy + extensions. So we only fill in what the server omitted. + """ + sep = b"\r\n\r\n" + if sep not in response: + return response + header_section, body = response.split(sep, 1) + lines = header_section.decode(errors="replace").split("\r\n") + + existing = {ln.split(":", 1)[0].strip().lower() + for ln in lines if ":" in ln} + + # If the upstream already handled CORS, leave it completely alone. + if "access-control-allow-origin" in existing: + return response + + # Otherwise inject a minimal, credential-safe set (no wildcards, + # since wildcards combined with credentials are invalid). + allow_origin = origin or "*" + additions = [f"Access-Control-Allow-Origin: {allow_origin}"] + if allow_origin != "*": + additions.append("Access-Control-Allow-Credentials: true") + additions.append("Vary: Origin") + return ("\r\n".join(lines + additions) + "\r\n\r\n").encode() + body + + async def _relay_smart(self, method, url, headers, body): + """Choose optimal relay strategy based on request type. + + - GET requests for likely-large downloads use parallel-range. + - All other requests (API calls, HTML, JSON, XHR) go through the + single-request relay. This avoids injecting a synthetic Range + header on normal traffic, which some origins honor by returning + 206 — breaking fetch()/XHR on sites like x.com or Cloudflare + challenge pages. + """ + if method == "GET" and not body: + # Respect client's own Range header verbatim. + if headers: + for k in headers: + if k.lower() == "range": + return await self.fronter.relay( + method, url, headers, body + ) + # Only probe with Range when the URL looks like a big file. + if self._is_likely_download(url, headers): + return await self.fronter.relay_parallel( + method, url, headers, body + ) + return await self.fronter.relay(method, url, headers, body) + + def _is_likely_download(self, url: str, headers: dict) -> bool: + """Heuristic: is this URL likely a large file download?""" + # Check file extension + path = url.split("?")[0].lower() + large_exts = { + ".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar", + ".exe", ".msi", ".dmg", ".deb", ".rpm", ".apk", + ".iso", ".img", + ".mp4", ".mkv", ".avi", ".mov", ".webm", + ".mp3", ".flac", ".wav", ".aac", + ".pdf", ".doc", ".docx", ".ppt", ".pptx", + ".wasm", + } + for ext in large_exts: + if path.endswith(ext): + return True + return False + + # ── Plain HTTP forwarding ───────────────────────────────────── + + async def _do_http(self, header_block: bytes, reader, writer): + body = b"" + for raw_line in header_block.split(b"\r\n"): + if raw_line.lower().startswith(b"content-length:"): + length = int(raw_line.split(b":", 1)[1].strip()) + body = await reader.readexactly(length) + break + + first_line = header_block.split(b"\r\n")[0].decode(errors="replace") + log.info("HTTP → %s", first_line) + + if self.mode == "apps_script": + # Parse request and relay through Apps Script + parts = first_line.strip().split(" ", 2) + method = parts[0] if parts else "GET" + url = parts[1] if len(parts) > 1 else "/" + + headers = {} + for raw_line in header_block.split(b"\r\n")[1:]: + if b":" in raw_line: + k, v = raw_line.decode(errors="replace").split(":", 1) + headers[k.strip()] = v.strip() + + # ── CORS preflight over plain HTTP ──────────────────────────── + origin = next( + (v for k, v in headers.items() if k.lower() == "origin"), "" + ) + acr_method = next( + (v for k, v in headers.items() + if k.lower() == "access-control-request-method"), "" + ) + acr_headers_val = next( + (v for k, v in headers.items() + if k.lower() == "access-control-request-headers"), "" + ) + if method.upper() == "OPTIONS" and acr_method: + log.debug("CORS preflight (HTTP) → %s (responding locally)", url[:60]) + writer.write(self._cors_preflight_response(origin, acr_method, acr_headers_val)) + await writer.drain() + return + + # Cache check for GET + response = None + if method == "GET" and not body: + response = self._cache.get(url) + if response: + log.debug("Cache HIT (HTTP): %s", url[:60]) + + if response is None: + response = await self._relay_smart(method, url, headers, body) + # Cache successful GET + if method == "GET" and not body and response: + ttl = ResponseCache.parse_ttl(response, url) + if ttl > 0: + self._cache.put(url, response, ttl) + + # Inject CORS headers for cross-origin requests + if origin and response: + response = self._inject_cors_headers(response, origin) + elif self.mode in ("google_fronting", "custom_domain", "domain_fronting"): + # Use WebSocket tunnel for ALL traffic (much faster than forward()) + response = await self._tunnel_http(header_block, body) + else: + response = await self.fronter.forward(header_block + body) + + writer.write(response) + await writer.drain() + + async def _tunnel_http(self, header_block: bytes, body: bytes) -> bytes: + """Forward plain HTTP via a persistent WebSocket tunnel. + + Instead of opening a new TLS+HTTP connection for each request + (the old forward() path), this keeps a WebSocket tunnel open + to the target host and pipes raw HTTP through it. + Much faster for rapid-fire requests (e.g., Telegram API). + """ + import re as _re + + # Parse target host:port from the raw HTTP request + host = "" + port = 80 + for line in header_block.split(b"\r\n")[1:]: + if not line: + break + if line.lower().startswith(b"host:"): + host_val = line.split(b":", 1)[1].strip().decode(errors="replace") + if ":" in host_val: + h, p = host_val.rsplit(":", 1) + try: + host, port = h, int(p) + except ValueError: + host = host_val + else: + host = host_val + break + + if not host: + return b"HTTP/1.1 400 Bad Request\r\n\r\nNo Host header\r\n" + + # Rewrite the request line: browser sends absolute URL + # (e.g., "GET http://host/path HTTP/1.1") but the target + # server expects a relative path ("GET /path HTTP/1.1") + first_line = header_block.split(b"\r\n")[0] + first_str = first_line.decode(errors="replace") + parts = first_str.split(" ", 2) + if len(parts) >= 2 and parts[1].startswith("http://"): + from urllib.parse import urlparse + parsed = urlparse(parts[1]) + rel_path = parsed.path or "/" + if parsed.query: + rel_path += "?" + parsed.query + new_first = f"{parts[0]} {rel_path}" + if len(parts) == 3: + new_first += f" {parts[2]}" + header_block = new_first.encode() + b"\r\n" + b"\r\n".join(header_block.split(b"\r\n")[1:]) + + raw_request = header_block + body + + # Send through tunnel + try: + return await asyncio.wait_for( + self.fronter.forward(raw_request), timeout=30 + ) + except Exception as e: + log.error("Tunnel HTTP failed (%s:%d): %s", host, port, e) + return b"HTTP/1.1 502 Bad Gateway\r\n\r\nTunnel forward failed\r\n" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5ff2a6a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +# Core (no external dependencies for basic modes) +# Python 3.10+ required + +# Optional: MITM interception (apps_script mode) +cryptography>=41.0.0 + +# Optional: HTTP/2 multiplexing (faster apps_script relay) +h2>=4.1.0 diff --git a/ws.py b/ws.py new file mode 100644 index 0000000..0c3ed22 --- /dev/null +++ b/ws.py @@ -0,0 +1,76 @@ +""" +Minimal WebSocket frame encoder / decoder (RFC 6455). + +Only handles binary (opcode 0x02) and close (opcode 0x08) frames. +Client-to-server frames are always masked as required by the spec. +""" + +import os +import struct + + +def ws_encode(data: bytes, opcode: int = 0x02) -> bytes: + """Encode *data* into a masked binary WebSocket frame.""" + head = bytearray([0x80 | opcode]) # FIN + opcode + + length = len(data) + if length < 126: + head.append(0x80 | length) + elif length < 0x10000: + head.append(0x80 | 126) + head += struct.pack("!H", length) + else: + head.append(0x80 | 127) + head += struct.pack("!Q", length) + + mask = os.urandom(4) + head += mask + + masked = bytearray(data) + for i in range(len(masked)): + masked[i] ^= mask[i & 3] + + return bytes(head) + bytes(masked) + + +def ws_decode(buf: bytes): + """Try to decode one frame from *buf*. + + Returns ``(opcode, payload, consumed_bytes)`` or ``None`` if the + buffer does not yet contain a complete frame. + """ + if len(buf) < 2: + return None + + opcode = buf[0] & 0x0F + is_masked = buf[1] & 0x80 + length = buf[1] & 0x7F + pos = 2 + + if length == 126: + if len(buf) < 4: + return None + length = struct.unpack("!H", buf[2:4])[0] + pos = 4 + elif length == 127: + if len(buf) < 10: + return None + length = struct.unpack("!Q", buf[2:10])[0] + pos = 10 + + mask = None + if is_masked: + if len(buf) < pos + 4: + return None + mask = buf[pos : pos + 4] + pos += 4 + + if len(buf) < pos + length: + return None + + payload = bytearray(buf[pos : pos + length]) + if mask: + for i in range(len(payload)): + payload[i] ^= mask[i & 3] + + return opcode, bytes(payload), pos + length