diff --git a/README.md b/README.md index 27478e7..c56d6c0 100644 --- a/README.md +++ b/README.md @@ -298,6 +298,7 @@ This project focuses entirely on the **Apps Script** relay — a free Google acc | `bypass_hosts` | `["localhost", ".local", ".lan", ".home.arpa"]` | Hosts that go direct (no MITM, no relay). Useful for LAN resources or sites that break under MITM. | | `direct_google_exclude` | see [config.example.json](config.example.json) | Google apps that must use the MITM relay path instead of the fast direct tunnel. | | `hosts` | `{}` | Manual DNS override: map a hostname to a specific IP. | +| `youtube_via_relay` | `false` | Route YouTube (`youtube.com`, `youtu.be`, `youtube-nocookie.com`) through the Apps Script relay instead of the SNI-rewrite path. The SNI-rewrite path uses Google's frontend IP which enforces SafeSearch and can cause **"Video Unavailable"** errors. Setting this to `true` fixes playback at the cost of using more Apps Script executions and slightly higher latency. | ### Optional Dependencies @@ -344,10 +345,53 @@ python3 main.py --log-level DEBUG # Show detailed logs python3 main.py -c /path/to/config.json # Use a different config file python3 main.py --install-cert # Install MITM CA certificate and exit python3 main.py --no-cert-check # Skip automatic CA install check on startup +python3 main.py --scan # Scan Google IPs and find the fastest one ``` > **Auto-install:** On startup (MITM mode), the proxy automatically checks if the CA certificate is trusted and attempts to install it. Use `--no-cert-check` to skip this. If auto-install fails (e.g. needs elevation), run `python main.py --install-cert` manually or follow Step 6 above. +### Scanning for the Fastest Google IP + +If your current `google_ip` in `config.json` is blocked or slow, you can scan to find a faster one: + +```bash +python3 main.py --scan +``` + +This will: +1. Probe 27 candidate Google IPs in parallel +2. Measure latency from your network +3. Display results in a table +4. Recommend the fastest IP +5. Exit with exit code 0 if at least one IP is reachable, 1 otherwise + +**Example output:** +``` +Scanning 27 Google frontend IPs + SNI: www.google.com + Timeout: 4s per IP + Concurrency: 8 parallel probes + +IP LATENCY STATUS +-------------------- ------------ ------------------------- +216.239.32.120 42ms OK +216.239.34.120 45ms OK +216.239.36.120 52ms OK +142.250.80.142 timeout timeout +... + +Result: 15 / 27 reachable + +Top 3 fastest IPs: + 1. 216.239.32.120 (42ms) + 2. 216.239.34.120 (45ms) + 3. 216.239.36.120 (52ms) + +Recommended: Set "google_ip": "216.239.32.120" in config.json +``` + +After scanning, update your `config.json` with the recommended IP and restart the proxy. + --- ## Architecture @@ -384,6 +428,7 @@ MasterHttpRelayVPN/ ├── mitm.py # On-the-fly TLS interception ├── cert_installer.py # Cross-platform CA installer (Windows/macOS/Linux + Firefox) ├── codec.py # Content-Encoding decoder (gzip/deflate/br/zstd) + ├── google_ip_scanner.py # Scanner to find the fastest reachable Google IP ├── constants.py # Tunable defaults and shared data └── logging_utils.py # Colored, aligned log formatter ``` diff --git a/README_FA.md b/README_FA.md index 09614db..b3f1845 100644 --- a/README_FA.md +++ b/README_FA.md @@ -24,7 +24,7 @@ `masterking32.ton` -- آدرس روی شبکه‌های EVM (ETH و سازگارها): +- آدرس روی شبکه‌های EVM (ETH و سازگارها): `0x517f07305D6ED781A089322B6cD93d1461bF8652` @@ -246,6 +246,7 @@ json | `block_hosts` | `[]` | هاست‌هایی که هرگز نباید tunnel شوند (پاسخ 403). نام دقیق (`ads.example.com`) یا پسوند با نقطه‌ی ابتدایی (`.doubleclick.net`). | | `bypass_hosts` | `["localhost", ".local", ".lan", ".home.arpa"]` | هاست‌هایی که مستقیم می‌روند (بدون MITM و بدون رله). برای منابع داخلی شبکه یا سایت‌هایی که با MITM مشکل دارند. | | `direct_google_exclude` | مراجعه به [config.example.json](config.example.json) | اپ‌های Google که باید از مسیر MITM برای رله استفاده کنند به‌جای tunnel مستقیم. | +| `youtube_via_relay` | `false` | مسیردهی YouTube (`youtube.com`، `youtu.be`، `youtube-nocookie.com`) از طریق رله Apps Script به‌جای مسیر SNI-rewrite. مسیر SNI-rewrite از IP فرانت‌اند Google عبور می‌کند که SafeSearch را اجباری می‌کند و می‌تواند باعث خطای **«ویدیو در دسترس نیست»** شود. با فعال کردن این گزینه، پخش ویدیو درست می‌شود اما تعداد اجراهای Apps Script بیشتر و تأخیر اندکی بالاتر می‌رود. | ### وابستگی‌های اختیاری @@ -291,10 +292,53 @@ python3 main.py --log-level DEBUG python3 main.py -c /path/to/config.json python3 main.py --install-cert # نصب گواهی CA و خروج python3 main.py --no-cert-check # رد شدن از بررسی خودکار گواهی +python3 main.py --scan # اسکن IP های Google و یافتن سریع‌ترین ``` > **نصب خودکار:** هنگام اجرا در حالت `apps_script`، برنامه به‌طور خودکار بررسی می‌کند که آیا گواهی CA قابل اعتماد است یا نه و در صورت نیاز آن را نصب می‌کند. اگر نصب خودکار ناموفق بود (مثلاً نیاز به دسترسی مدیر دارد)، می‌توانید دستور `python main.py --install-cert` را اجرا کنید یا مراحل مرحله ۶ را دنبال کنید. +### اسکن کردن برای یافتن سریع‌ترین IP گوگل + +اگر `google_ip` فعلی در `config.json` بلاک شده یا آهسته است، می‌توانید اسکن کنید تا سریع‌ترین آن را پیدا کنید: + +```bash +python3 main.py --scan +``` + +این دستور: +1. ۲۷ IP برای fronting Google را به‌صورت موازی بررسی می‌کند +2. تأخیر (latency) از شبکه شما را اندازه می‌گیرد +3. نتایج را در جدول نمایش می‌دهد +4. سریع‌ترین IP را پیشنهاد می‌دهد +5. اگر حداقل یک IP در دسترس باشد کد خروج ۰، ورنه ۱ را برمی‌گرداند + +**نمونه خروجی:** +``` +Scanning 27 Google frontend IPs + SNI: www.google.com + Timeout: 4s per IP + Concurrency: 8 parallel probes + +IP LATENCY STATUS +-------------------- ------------ ------------------------- +216.239.32.120 42ms OK +216.239.34.120 45ms OK +216.239.36.120 52ms OK +142.250.80.142 timeout timeout +... + +Result: 15 / 27 reachable + +Top 3 fastest IPs: + 1. 216.239.32.120 (42ms) + 2. 216.239.34.120 (45ms) + 3. 216.239.36.120 (52ms) + +Recommended: Set "google_ip": "216.239.32.120" in config.json +``` + +پس از اسکن، مقدار `google_ip` در `config.json` را با IP پیشنهادی به‌روزرسانی کنید و پراکسی را دوباره راه‌اندازی کنید. + --- ## معماری @@ -327,6 +371,7 @@ MasterHttpRelayVPN/ ├── mitm.py # ساخت و مدیریت گواهی‌ها ├── cert_installer.py # نصب خودکار CA در ویندوز/مک/لینوکس + فایرفاکس ├── codec.py # رمزگشای Content-Encoding (gzip/deflate/br/zstd) + ├── google_ip_scanner.py # اسکنر IP های Google برای یافتن سریع‌ترین ├── constants.py # مقادیر پیش‌فرض قابل تنظیم └── logging_utils.py # فرمت‌دهنده‌ی لاگ رنگی و منظم ``` diff --git a/config.example.json b/config.example.json index 639aae7..149434a 100644 --- a/config.example.json +++ b/config.example.json @@ -83,5 +83,6 @@ "www.google.com", "safebrowsing.google.com" ], + "youtube_via_relay": false, "hosts": {} } diff --git a/main.py b/main.py index 1aea240..1fb76a8 100644 --- a/main.py +++ b/main.py @@ -23,6 +23,7 @@ if _SRC_DIR not in sys.path: from cert_installer import install_ca, is_ca_trusted from constants import __version__ from lan_utils import log_lan_access +from google_ip_scanner import scan_sync from logging_utils import configure as configure_logging, print_banner from mitm import CA_CERT_FILE from proxy_server import ProxyServer @@ -92,6 +93,11 @@ def parse_args(): action="store_true", help="Skip the certificate installation check on startup.", ) + parser.add_argument( + "--scan", + action="store_true", + help="Scan Google IPs to find the fastest reachable one and exit.", + ) return parser.parse_args() @@ -192,6 +198,15 @@ def main(): ok = install_ca(CA_CERT_FILE) sys.exit(0 if ok else 1) + # ── Google IP Scanner ────────────────────────────────────────────────── + if args.scan: + setup_logging("INFO") + front_domain = config.get("front_domain", "www.google.com") + _log = logging.getLogger("Main") + _log.info(f"Scanning Google IPs (fronting domain: {front_domain})") + ok = scan_sync(front_domain) + sys.exit(0 if ok else 1) + setup_logging(config.get("log_level", "INFO")) log = logging.getLogger("Main") diff --git a/requirements.txt b/requirements.txt index ee2b572..5b0e2a3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,6 +7,9 @@ cryptography>=41.0.0 # Optional: HTTP/2 multiplexing (faster apps_script relay) h2>=4.1.0 +# Optional: CA bundle for TLS verification (recommended on macOS / Windows) +certifi>=2024.1.0 + # Optional: Brotli decompression (modern websites send `br` encoding) brotli>=1.1.0 diff --git a/src/constants.py b/src/constants.py index 5a1810c..80cf5c4 100644 --- a/src/constants.py +++ b/src/constants.py @@ -23,6 +23,39 @@ RELAY_TIMEOUT = 25 TLS_CONNECT_TIMEOUT = 15 TCP_CONNECT_TIMEOUT = 10 +# ── Google IP Scanner settings ────────────────────────────────────────────── +GOOGLE_SCANNER_TIMEOUT = 4 # Timeout per IP probe (seconds) +GOOGLE_SCANNER_CONCURRENCY = 8 # Parallel probes +# Candidate Google frontend IPs for scanning (multiple ASNs and regions) +CANDIDATE_IPS: tuple[str, ...] = ( + "216.239.32.120", + "216.239.34.120", + "216.239.36.120", + "216.239.38.120", + "142.250.80.142", + "142.250.80.138", + "142.250.179.110", + "142.250.185.110", + "142.250.184.206", + "142.250.190.238", + "142.250.191.78", + "172.217.1.206", + "172.217.14.206", + "172.217.16.142", + "172.217.22.174", + "172.217.164.110", + "172.217.168.206", + "172.217.169.206", + "34.107.221.82", + "142.251.32.110", + "142.251.33.110", + "142.251.46.206", + "142.251.46.238", + "142.250.80.170", + "142.250.72.206", + "142.250.64.206", + "142.250.72.110", +) # ── Response cache ──────────────────────────────────────────────────────── CACHE_MAX_MB = 50 diff --git a/src/domain_fronter.py b/src/domain_fronter.py index f2b1084..15007ec 100644 --- a/src/domain_fronter.py +++ b/src/domain_fronter.py @@ -21,6 +21,11 @@ import time from dataclasses import dataclass from urllib.parse import urlparse +try: + import certifi +except Exception: # optional dependency fallback + certifi = None + import codec from constants import ( BATCH_MAX, @@ -167,6 +172,8 @@ class DomainFronter: self._batch_window_macro = BATCH_WINDOW_MACRO self._batch_max = BATCH_MAX self._batch_enabled = True + self._batch_disabled_at = 0.0 + self._batch_cooldown = 60 # Request coalescing — dedup concurrent identical GETs self._coalesce: dict[str, list[asyncio.Future]] = {} @@ -219,6 +226,11 @@ class DomainFronter: def _ssl_ctx(self) -> ssl.SSLContext: ctx = ssl.create_default_context() + if certifi is not None: + try: + ctx.load_verify_locations(cafile=certifi.where()) + except Exception: + pass if not self.verify_ssl: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE @@ -281,7 +293,7 @@ class DomainFronter: we rotate across `self._sni_hosts` so DPI can't fingerprint "always www.google.com" from the client side. """ - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.setblocking(False) @@ -307,7 +319,7 @@ class DomainFronter: async def _acquire(self): """Get a healthy TLS connection from pool (TTL-checked) or open new.""" - now = asyncio.get_event_loop().time() + now = asyncio.get_running_loop().time() async with self._pool_lock: while self._pool: reader, writer, created = self._pool.pop() @@ -326,11 +338,11 @@ class DomainFronter: if not self._refilling: self._refilling = True self._spawn(self._refill_pool()) - return reader, writer, asyncio.get_event_loop().time() + return reader, writer, asyncio.get_running_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() + now = asyncio.get_running_loop().time() if (now - created) >= self._conn_ttl or reader.at_eof(): try: writer.close() @@ -708,7 +720,7 @@ class DomainFronter: """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() + t = asyncio.get_running_loop().time() async with self._pool_lock: if len(self._pool) < self._pool_max: self._pool.append((r, w, t)) @@ -725,7 +737,7 @@ class DomainFronter: while True: try: await asyncio.sleep(3) - now = asyncio.get_event_loop().time() + now = asyncio.get_running_loop().time() # Purge expired / dead connections async with self._pool_lock: @@ -973,7 +985,7 @@ class DomainFronter: race where the owning task's `finally` pops the entry between the check and append by a second task. """ - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() async with self._batch_lock: waiters = self._coalesce.get(key) if waiters is not None: @@ -1152,12 +1164,12 @@ class DomainFronter: f"chunk {s}-{e} failed after {max_tries} tries: {last_err}" ) - t0 = asyncio.get_event_loop().time() + t0 = asyncio.get_running_loop().time() results = await asyncio.gather( *[fetch_range(s, e) for s, e in ranges], return_exceptions=True, ) - elapsed = asyncio.get_event_loop().time() - t0 + elapsed = asyncio.get_running_loop().time() - t0 # Assemble full body parts = [resp_body] @@ -1498,11 +1510,21 @@ class DomainFronter: async def _batch_submit(self, payload: dict) -> bytes: """Submit a request to the batch collector. Returns raw HTTP response.""" - # If batching is disabled (old Code.gs), go direct + # If batching is disabled, retry enabling it after a cooldown. if not self._batch_enabled: - return await self._relay_with_retry(payload) + if ( + self._batch_disabled_at > 0 + and (time.time() - self._batch_disabled_at) >= self._batch_cooldown + ): + self._batch_enabled = True + log.info( + "Batch mode re-enabled after %ds cooldown", + self._batch_cooldown, + ) + else: + return await self._relay_with_retry(payload) - future = asyncio.get_event_loop().create_future() + future = asyncio.get_running_loop().create_future() async with self._batch_lock: self._batch_pending.append((payload, future)) @@ -1568,9 +1590,13 @@ class DomainFronter: if not future.done(): future.set_result(result) except Exception as e: - log.warning("Batch relay failed, disabling batch mode. " - "Redeploy Code.gs for batch support. Error: %s", e) + log.warning( + "Batch relay failed, disabling batch mode for %ds cooldown. " + "Error: %s", + self._batch_cooldown, e, + ) self._batch_enabled = False + self._batch_disabled_at = time.time() # Fallback: send individually tasks = [] for payload, future in batch: diff --git a/src/google_ip_scanner.py b/src/google_ip_scanner.py new file mode 100644 index 0000000..79edc51 --- /dev/null +++ b/src/google_ip_scanner.py @@ -0,0 +1,194 @@ +""" +Google IP Scanner — finds the fastest reachable Google frontend IP. + +Scans a list of candidate Google IPs via HTTPS (with SNI fronting), measures +latency, and reports results in a formatted table. Useful for finding the best +IP to configure in config.json when your current IP is blocked. +""" + +from __future__ import annotations + +import asyncio +import logging +import ssl +import time +from dataclasses import dataclass +from typing import Optional + +from constants import CANDIDATE_IPS, GOOGLE_SCANNER_TIMEOUT, GOOGLE_SCANNER_CONCURRENCY + +log = logging.getLogger("Scanner") + + +@dataclass +class ProbeResult: + """Result of a single IP probe.""" + ip: str + latency_ms: Optional[int] = None + error: Optional[str] = None + + @property + def ok(self) -> bool: + return self.latency_ms is not None + + +async def _probe_ip( + ip: str, + sni: str, + semaphore: asyncio.Semaphore, + timeout: float, +) -> ProbeResult: + """ + Probe a single IP via HTTPS with SNI fronting. + + Args: + ip: The IP to probe (xxx.xxx.xxx.xxx). + sni: The SNI hostname to use in TLS handshake. + semaphore: Rate limiter to control concurrency. + timeout: Timeout in seconds for the entire probe. + + Returns: + ProbeResult with latency_ms (if successful) or error message. + """ + async with semaphore: + start_time = time.time() + try: + # Create SSL context that skips certificate verification + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + + # Connect to IP:443 with SNI set to the fronting domain + reader, writer = await asyncio.wait_for( + asyncio.open_connection( + ip, + 443, + ssl=ctx, + server_hostname=sni, + ), + timeout=timeout, + ) + + # Send minimal HTTP HEAD request + request = f"HEAD / HTTP/1.1\r\nHost: {sni}\r\nConnection: close\r\n\r\n" + writer.write(request.encode()) + await writer.drain() + + # Read response header (first 256 bytes is plenty for HTTP status) + response = await asyncio.wait_for(reader.read(256), timeout=timeout) + + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + # Check if we got an HTTP response + if not response: + return ProbeResult(ip=ip, error="empty response") + + response_str = response.decode("utf-8", errors="ignore") + if not response_str.startswith("HTTP/"): + return ProbeResult(ip=ip, error=f"invalid response: {response_str[:30]!r}") + + # Success — return latency in milliseconds + elapsed_ms = int((time.time() - start_time) * 1000) + return ProbeResult(ip=ip, latency_ms=elapsed_ms) + + except asyncio.TimeoutError: + return ProbeResult(ip=ip, error="timeout") + except ConnectionRefusedError: + return ProbeResult(ip=ip, error="connection refused") + except ConnectionResetError: + return ProbeResult(ip=ip, error="connection reset") + except OSError as e: + return ProbeResult(ip=ip, error=f"network error: {e.strerror or str(e)}") + except Exception as e: + return ProbeResult(ip=ip, error=f"probe failed: {type(e).__name__}") + + +async def run(front_domain: str) -> bool: + """ + Scan all candidate Google IPs and display results. + + Args: + front_domain: The SNI hostname to use (e.g. "www.google.com"). + + Returns: + True if at least one IP is reachable, False otherwise. + """ + timeout = GOOGLE_SCANNER_TIMEOUT + concurrency = GOOGLE_SCANNER_CONCURRENCY + + print() + print(f"Scanning {len(CANDIDATE_IPS)} Google frontend IPs") + print(f" SNI: {front_domain}") + print(f" Timeout: {timeout}s per IP") + print(f" Concurrency: {concurrency} parallel probes") + print() + + # Create semaphore to limit concurrency + semaphore = asyncio.Semaphore(concurrency) + + # Launch all probes concurrently + tasks = [ + _probe_ip(ip, front_domain, semaphore, timeout) + for ip in CANDIDATE_IPS + ] + results = await asyncio.gather(*tasks) + + # Sort by latency (successful first, then by speed) + results.sort(key=lambda r: (not r.ok, r.latency_ms or float("inf"))) + + # Display results table + print(f"{'IP':<20} {'LATENCY':<12} {'STATUS':<25}") + print(f"{'-' * 20} {'-' * 12} {'-' * 25}") + + ok_count = 0 + for result in results: + if result.ok: + print(f"{result.ip:<20} {result.latency_ms:>8}ms OK") + ok_count += 1 + else: + status = result.error or "unknown error" + print(f"{result.ip:<20} {'—':<12} {status:<25}") + + print() + print(f"Result: {ok_count} / {len(results)} reachable") + + if ok_count == 0: + print("No Google IPs reachable from this network.") + print() + return False + + # Show top 3 fastest + fastest = [r for r in results if r.ok][:3] + print() + print("Top 3 fastest IPs:") + for i, result in enumerate(fastest, 1): + print(f" {i}. {result.ip} ({result.latency_ms}ms)") + + print() + print(f"Recommended: Set \"google_ip\": \"{fastest[0].ip}\" in config.json") + print() + return True + + +def scan_sync(front_domain: str) -> bool: + """ + Wrapper to run async scanner from sync context (e.g. main.py). + + Args: + front_domain: The SNI hostname to use. + + Returns: + True if at least one IP is reachable, False otherwise. + """ + try: + return asyncio.run(run(front_domain)) + except KeyboardInterrupt: + print("\nScan interrupted by user.") + return False + except Exception as e: + log.error(f"Scan failed: {e}") + return False diff --git a/src/h2_transport.py b/src/h2_transport.py index edeb1d1..82c60ee 100644 --- a/src/h2_transport.py +++ b/src/h2_transport.py @@ -20,6 +20,11 @@ import socket import ssl from urllib.parse import urlparse +try: + import certifi +except Exception: # optional dependency fallback + certifi = None + import codec log = logging.getLogger("H2") @@ -107,6 +112,13 @@ class H2Transport: async def _do_connect(self): """Establish the HTTP/2 connection with optimized socket settings.""" ctx = ssl.create_default_context() + # Some Python builds don't expose a usable default CA store. + # Load certifi bundle when present to keep TLS verification stable. + if certifi is not None: + try: + ctx.load_verify_locations(cafile=certifi.where()) + except Exception: + pass # 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: @@ -128,7 +140,7 @@ class H2Transport: try: await asyncio.wait_for( - asyncio.get_event_loop().sock_connect( + asyncio.get_running_loop().sock_connect( raw, (self.connect_host, 443) ), timeout=15, @@ -360,6 +372,15 @@ class H2Transport: except asyncio.CancelledError: pass + except ssl.SSLError as e: + # APPLICATION_DATA_AFTER_CLOSE_NOTIFY is raised when the server + # sends data after its TLS close_notify — technically a protocol + # violation but very common with CDNs. It just means the + # connection is closed; reconnect on the next request. + if "APPLICATION_DATA_AFTER_CLOSE_NOTIFY" in str(e): + log.debug("H2 TLS session closed by remote (close_notify): %s", e) + else: + log.error("H2 reader error: %s", e) except Exception as e: if "application data after close notify" in str(e).lower(): log.debug("H2 reader closed after close_notify: %s", e) diff --git a/src/proxy_server.py b/src/proxy_server.py index cbf4d46..37cab79 100644 --- a/src/proxy_server.py +++ b/src/proxy_server.py @@ -15,6 +15,11 @@ import time import ipaddress from urllib.parse import urlparse +try: + import certifi +except Exception: # optional dependency fallback + certifi = None + from constants import ( CACHE_MAX_MB, CACHE_TTL_MAX, @@ -237,6 +242,17 @@ class ProxyServer: self._block_hosts = self._load_host_rules(config.get("block_hosts", [])) self._bypass_hosts = self._load_host_rules(config.get("bypass_hosts", [])) + # Route YouTube through the relay when requested; the Google frontend + # IP can enforce SafeSearch on the SNI-rewrite path. + if config.get("youtube_via_relay", False): + self._SNI_REWRITE_SUFFIXES = tuple( + s for s in SNI_REWRITE_SUFFIXES + if s not in self._YOUTUBE_SNI_SUFFIXES + ) + log.info("youtube_via_relay enabled — YouTube routed through relay") + else: + self._SNI_REWRITE_SUFFIXES = SNI_REWRITE_SUFFIXES + try: from mitm import MITMCertManager self.mitm = MITMCertManager() @@ -739,6 +755,11 @@ class ProxyServer: # Built-in list of domains that must be reached via Google's frontend IP # with SNI rewritten to `front_domain` (default: www.google.com). # Source: constants.SNI_REWRITE_SUFFIXES. + # When youtube_via_relay is enabled the YouTube suffixes are removed so + # YouTube goes through the Apps Script relay instead. + _YOUTUBE_SNI_SUFFIXES = frozenset({ + "youtube.com", "youtu.be", "youtube-nocookie.com", + }) _SNI_REWRITE_SUFFIXES = SNI_REWRITE_SUFFIXES def _sni_rewrite_ip(self, host: str) -> str | None: @@ -973,7 +994,7 @@ class ProxyServer: # Step 1: MITM — accept TLS from the browser ssl_ctx_server = self.mitm.get_server_context(host) - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() transport = writer.transport protocol = transport.get_protocol() try: @@ -987,6 +1008,11 @@ class ProxyServer: # Step 2: open outgoing TLS to target IP with the safe SNI ssl_ctx_client = ssl.create_default_context() + if certifi is not None: + try: + ssl_ctx_client.load_verify_locations(cafile=certifi.where()) + except Exception: + pass if not self.fronter.verify_ssl: ssl_ctx_client.check_hostname = False ssl_ctx_client.verify_mode = ssl.CERT_NONE @@ -1040,7 +1066,7 @@ class ProxyServer: ssl_ctx = self.mitm.get_server_context(host) # Upgrade the existing connection to TLS (we are the server) - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() transport = writer.transport protocol = transport.get_protocol()