Improve apps_script relay stability and add SOCKS5 support

config.example.json

@@ -7,8 +7,15 @@
 	"auth_key": "CHANGE_ME_TO_A_STRONG_SECRET",
 	"listen_host": "127.0.0.1",
 	"listen_port": 8085,
+	"socks5_enabled": true,
+	"socks5_host": "127.0.0.1",
+	"socks5_port": 1080,
 	"log_level": "INFO",
 	"verify_ssl": true,
+	"_direct_google_exclude_comment": "Google web apps that should NEVER use the raw direct-tunnel shortcut. Supports exact hosts and optional suffix patterns like \".googleapis.com\". They will go through the MITM relay path instead for better compatibility.",
+	"direct_google_exclude": ["gemini.google.com", "aistudio.google.com", "notebooklm.google.com", "labs.google.com", "meet.google.com", "accounts.google.com", "ogs.google.com", "mail.google.com", "calendar.google.com", "drive.google.com", "docs.google.com", "chat.google.com"],
+	"_direct_google_allow_comment": "Conservative allowlist for raw direct Google tunneling. Leave empty unless you have confirmed a host works better direct than via relay.",
+	"direct_google_allow": ["www.google.com", "safebrowsing.google.com"],
 	"_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": {}
 }

domain_fronter.py

@@ -19,6 +19,7 @@ Mode 4 (apps_script):
 
 import asyncio
 import base64
+import hashlib
 import gzip
 import json
 import logging
@@ -34,6 +35,12 @@ log = logging.getLogger("Fronter")
 
 
 class DomainFronter:
+    _STATIC_EXTS = (
+        ".css", ".js", ".mjs", ".woff", ".woff2", ".ttf", ".eot",
+        ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".ico",
+        ".mp3", ".mp4", ".webm", ".wasm", ".avif",
+    )
+
     def __init__(self, config: dict):
         mode = config.get("mode", "domain_fronting")
 
@@ -170,9 +177,34 @@ class DomainFronter:
         self._script_idx += 1
         return sid
 
-    def _exec_path(self) -> str:
+    @staticmethod
-        """Get the next Apps Script endpoint path (/dev or /exec)."""
+    def _host_key(url_or_host: str | None) -> str:
-        sid = self._next_script_id()
+        """Return a stable routing key for a URL or host string."""
+        if not url_or_host:
+            return ""
+        parsed = urlparse(url_or_host if "://" in url_or_host else f"https://{url_or_host}")
+        host = parsed.hostname or url_or_host
+        return host.lower().rstrip(".")
+
+    def _script_id_for_key(self, key: str | None = None) -> str:
+        """Pick a stable Apps Script ID for a host or fallback to round-robin.
+
+        When multiple deployments are configured, using a stable mapping per
+        host reduces IP/session churn for sites that are sensitive to endpoint
+        changes. If no key is available, we keep the older round-robin fallback
+        so warmup/keepalive traffic still distributes normally.
+        """
+        if len(self._script_ids) == 1:
+            return self._script_ids[0]
+        if not key:
+            return self._next_script_id()
+        digest = hashlib.sha1(key.encode("utf-8")).digest()
+        idx = int.from_bytes(digest[:4], "big") % len(self._script_ids)
+        return self._script_ids[idx]
+
+    def _exec_path(self, url_or_host: str | None = None) -> str:
+        """Get the Apps Script endpoint path (/dev or /exec)."""
+        sid = self._script_id_for_key(self._host_key(url_or_host))
         return f"/macros/s/{sid}/{'dev' if self._dev_available else 'exec'}"
 
     async def _flush_pool(self):
@@ -332,7 +364,7 @@ class DomainFronter:
 
                 # Apps Script keepalive — warm the container
                 payload = {"m": "HEAD", "u": "http://example.com/", "k": self.auth_key}
-                path = self._exec_path()
+                path = self._exec_path("example.com")
                 t0 = time.perf_counter()
                 await asyncio.wait_for(
                     self._h2.request(
@@ -514,6 +546,11 @@ class DomainFronter:
 
         payload = self._build_payload(method, url, headers, body)
 
+        # Stateful/browser-navigation requests should preserve exact ordering
+        # and header context; batching/coalescing is reserved for static fetches.
+        if self._is_stateful_request(method, url, headers, body):
+            return await self._relay_with_retry(payload)
+
         # 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.
@@ -711,7 +748,8 @@ class DomainFronter:
         payload = {
             "m": method,
             "u": url,
-            "r": True,
+            # Let the browser/app see origin redirects and cookies directly.
+            "r": False,
         }
         if headers:
             # Strip Accept-Encoding: Apps Script auto-decompresses gzip
@@ -726,6 +764,46 @@ class DomainFronter:
                 payload["ct"] = ct
         return payload
 
+    @classmethod
+    def _is_static_asset_url(cls, url: str) -> bool:
+        path = urlparse(url).path.lower()
+        return any(path.endswith(ext) for ext in cls._STATIC_EXTS)
+
+    @staticmethod
+    def _header_value(headers: dict | None, name: str) -> str:
+        if not headers:
+            return ""
+        for key, value in headers.items():
+            if key.lower() == name:
+                return str(value)
+        return ""
+
+    @classmethod
+    def _is_stateful_request(cls, method: str, url: str,
+                             headers: dict | None, body: bytes) -> bool:
+        method = method.upper()
+        if method not in {"GET", "HEAD"} or body:
+            return True
+
+        if headers:
+            for name in (
+                "cookie", "authorization", "proxy-authorization",
+                "origin", "referer", "if-none-match", "if-modified-since",
+                "cache-control", "pragma",
+            ):
+                if cls._header_value(headers, name):
+                    return True
+
+            accept = cls._header_value(headers, "accept").lower()
+            if "text/html" in accept or "application/json" in accept:
+                return True
+
+            fetch_mode = cls._header_value(headers, "sec-fetch-mode").lower()
+            if fetch_mode in {"navigate", "cors"}:
+                return True
+
+        return not cls._is_static_asset_url(url)
+
     # ── Batch collector ───────────────────────────────────────────
 
     async def _batch_submit(self, payload: dict) -> bytes:
@@ -866,7 +944,7 @@ class DomainFronter:
         full_payload["k"] = self.auth_key
         json_body = json.dumps(full_payload).encode()
 
-        path = self._exec_path()
+        path = self._exec_path(payload.get("u"))
 
         status, headers, body = await self._h2.request(
             method="POST", path=path, host=self.http_host,
@@ -883,7 +961,7 @@ class DomainFronter:
         full_payload["k"] = self.auth_key
         json_body = json.dumps(full_payload).encode()
 
-        path = self._exec_path()
+        path = self._exec_path(payload.get("u"))
         reader, writer, created = await self._acquire()
 
         try:
@@ -911,14 +989,22 @@ class DomainFronter:
 
                 parsed = urlparse(location)
                 rpath = parsed.path + ("?" + parsed.query if parsed.query else "")
-                request = (
+                if status in (307, 308):
-                    f"GET {rpath} HTTP/1.1\r\n"
+                    redirect_method = "POST"
-                    f"Host: {parsed.netloc}\r\n"
+                    redirect_body = json_body
-                    f"Accept-Encoding: gzip\r\n"
+                else:
-                    f"Connection: keep-alive\r\n"
+                    redirect_method = "GET"
-                    f"\r\n"
+                    redirect_body = b""
-                )
+                request_lines = [
-                writer.write(request.encode())
+                    f"{redirect_method} {rpath} HTTP/1.1",
+                    f"Host: {parsed.netloc}",
+                    "Accept-Encoding: gzip",
+                    "Connection: keep-alive",
+                ]
+                if redirect_body:
+                    request_lines.append(f"Content-Length: {len(redirect_body)}")
+                request = "\r\n".join(request_lines) + "\r\n\r\n"
+                writer.write(request.encode() + redirect_body)
                 await writer.drain()
                 status, resp_headers, resp_body = await self._read_http_response(reader)
 
@@ -939,7 +1025,7 @@ class DomainFronter:
             "q": payloads,
         }
         json_body = json.dumps(batch_payload).encode()
-        path = self._exec_path()
+        path = self._exec_path(payloads[0].get("u") if payloads else None)
 
         # Try HTTP/2 first
         if self._h2 and self._h2.is_connected:
@@ -983,14 +1069,22 @@ class DomainFronter:
                         break
                     parsed = urlparse(location)
                     rpath = parsed.path + ("?" + parsed.query if parsed.query else "")
-                    request = (
+                    if status in (307, 308):
-                        f"GET {rpath} HTTP/1.1\r\n"
+                        redirect_method = "POST"
-                        f"Host: {parsed.netloc}\r\n"
+                        redirect_body = json_body
-                        f"Accept-Encoding: gzip\r\n"
+                    else:
-                        f"Connection: keep-alive\r\n"
+                        redirect_method = "GET"
-                        f"\r\n"
+                        redirect_body = b""
-                    )
+                    request_lines = [
-                    writer.write(request.encode())
+                        f"{redirect_method} {rpath} HTTP/1.1",
+                        f"Host: {parsed.netloc}",
+                        "Accept-Encoding: gzip",
+                        "Connection: keep-alive",
+                    ]
+                    if redirect_body:
+                        request_lines.append(f"Content-Length: {len(redirect_body)}")
+                    request = "\r\n".join(request_lines) + "\r\n\r\n"
+                    writer.write(request.encode() + redirect_body)
                     await writer.drain()
                     status, resp_headers, resp_body = await self._read_http_response(reader)
 

main.py

@@ -51,6 +51,17 @@ def parse_args():
         default=None,
         help="Override listen host (env: DFT_HOST)",
     )
+    parser.add_argument(
+        "--socks5-port",
+        type=int,
+        default=None,
+        help="Override SOCKS5 listen port (env: DFT_SOCKS5_PORT)",
+    )
+    parser.add_argument(
+        "--disable-socks5",
+        action="store_true",
+        help="Disable the built-in SOCKS5 listener.",
+    )
     parser.add_argument(
         "--log-level",
         choices=["DEBUG", "INFO", "WARNING", "ERROR"],
@@ -107,6 +118,14 @@ def main():
     elif os.environ.get("DFT_HOST"):
         config["listen_host"] = os.environ["DFT_HOST"]
 
+    if args.socks5_port is not None:
+        config["socks5_port"] = args.socks5_port
+    elif os.environ.get("DFT_SOCKS5_PORT"):
+        config["socks5_port"] = int(os.environ["DFT_SOCKS5_PORT"])
+
+    if args.disable_socks5:
+        config["socks5_enabled"] = False
+
     if args.log_level is not None:
         config["log_level"] = args.log_level
     elif os.environ.get("DFT_LOG_LEVEL"):
@@ -162,7 +181,7 @@ def main():
                  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))
+            log.info("Script IDs        : %d scripts (sticky per-host)", len(script_ids))
             for i, sid in enumerate(script_ids):
                 log.info("  [%d] %s", i + 1, sid)
         else:
@@ -192,7 +211,13 @@ def main():
         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))
+    log.info("HTTP proxy         : %s:%d",
+             config.get("listen_host", "127.0.0.1"),
+             config.get("listen_port", 8080))
+    if config.get("socks5_enabled", True):
+        log.info("SOCKS5 proxy       : %s:%d",
+                 config.get("socks5_host", config.get("listen_host", "127.0.0.1")),
+                 config.get("socks5_port", 1080))
 
     try:
         asyncio.run(ProxyServer(config).start())

proxy_server.py

@@ -12,8 +12,11 @@ Supports:
 import asyncio
 import logging
 import re
+import socket
 import ssl
 import time
+import ipaddress
+from urllib.parse import urlparse
 
 from domain_fronter import DomainFronter
 
@@ -69,7 +72,7 @@ class ResponseCache:
         # 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:
+        if "no-store" in hdr or "private" in hdr or "set-cookie:" in hdr:
             return 0
 
         # Explicit max-age
@@ -101,13 +104,57 @@ class ResponseCache:
 
 
 class ProxyServer:
+    _GOOGLE_DIRECT_EXACT_EXCLUDE = {
+        "gemini.google.com",
+        "aistudio.google.com",
+        "notebooklm.google.com",
+        "labs.google.com",
+        "meet.google.com",
+        "accounts.google.com",
+        "ogs.google.com",
+        "mail.google.com",
+        "calendar.google.com",
+        "drive.google.com",
+        "docs.google.com",
+        "chat.google.com",
+        "photos.google.com",
+        "maps.google.com",
+        "myaccount.google.com",
+        "contacts.google.com",
+        "classroom.google.com",
+        "keep.google.com",
+        "play.google.com",
+    }
+    _GOOGLE_DIRECT_SUFFIX_EXCLUDE = (
+        ".meet.google.com",
+    )
+    _GOOGLE_DIRECT_ALLOW_EXACT = {
+        "www.google.com",
+        "google.com",
+        "safebrowsing.google.com",
+    }
+    _GOOGLE_DIRECT_ALLOW_SUFFIXES = ()
+    _TRACE_HOST_SUFFIXES = (
+        "chatgpt.com",
+        "openai.com",
+        "gemini.google.com",
+        "google.com",
+        "cloudflare.com",
+        "challenges.cloudflare.com",
+        "turnstile",
+    )
+
     def __init__(self, config: dict):
         self.host = config.get("listen_host", "127.0.0.1")
         self.port = config.get("listen_port", 8080)
+        self.socks_enabled = config.get("socks5_enabled", True)
+        self.socks_host = config.get("socks5_host", self.host)
+        self.socks_port = config.get("socks5_port", 1080)
         self.mode = config.get("mode", "domain_fronting")
         self.fronter = DomainFronter(config)
         self.mitm = None
         self._cache = ResponseCache(max_mb=50)
+        self._direct_fail_until: dict[str, float] = {}
 
         # Persistent HTTP tunnel cache for google_fronting mode
         # Key: "host:port" → (tunnel_reader, tunnel_writer, lock)
@@ -117,6 +164,22 @@ class ProxyServer:
         # 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", {})
+        configured_direct_exclude = config.get("direct_google_exclude", [])
+        self._direct_google_exclude = {
+            h.lower().rstrip(".")
+            for h in (
+                list(self._GOOGLE_DIRECT_EXACT_EXCLUDE) +
+                list(configured_direct_exclude)
+            )
+        }
+        configured_direct_allow = config.get("direct_google_allow", [])
+        self._direct_google_allow = {
+            h.lower().rstrip(".")
+            for h in (
+                list(self._GOOGLE_DIRECT_ALLOW_EXACT) +
+                list(configured_direct_allow)
+            )
+        }
 
         if self.mode == "apps_script":
             try:
@@ -127,14 +190,94 @@ class ProxyServer:
                 log.error("Run: pip install cryptography")
                 raise SystemExit(1)
 
+    @staticmethod
+    def _header_value(headers: dict | None, name: str) -> str:
+        if not headers:
+            return ""
+        for key, value in headers.items():
+            if key.lower() == name:
+                return str(value)
+        return ""
+
+    def _cache_allowed(self, method: str, url: str,
+                       headers: dict | None, body: bytes) -> bool:
+        if method.upper() != "GET" or body:
+            return False
+        for name in (
+            "cookie", "authorization", "proxy-authorization", "range",
+            "if-none-match", "if-modified-since", "cache-control", "pragma",
+        ):
+            if self._header_value(headers, name):
+                return False
+        return self.fronter._is_static_asset_url(url)
+
+    @classmethod
+    def _should_trace_host(cls, host: str) -> bool:
+        h = host.lower().rstrip(".")
+        return any(
+            token == h or token in h or h.endswith("." + token)
+            for token in cls._TRACE_HOST_SUFFIXES
+        )
+
+    def _log_response_summary(self, url: str, response: bytes):
+        status, headers, body = self.fronter._split_raw_response(response)
+        host = (urlparse(url).hostname or "").lower()
+        if status >= 300 or self._should_trace_host(host):
+            location = headers.get("location", "")
+            server = headers.get("server", "")
+            cf_ray = headers.get("cf-ray", "")
+            content_type = headers.get("content-type", "")
+            body_len = len(body)
+            body_hint = "-"
+            if "text/html" in content_type.lower() and body:
+                sample = body[:800].decode(errors="replace").lower()
+                if "<title>" in sample and "</title>" in sample:
+                    title = sample.split("<title>", 1)[1].split("</title>", 1)[0]
+                    body_hint = title[:120]
+                elif "captcha" in sample:
+                    body_hint = "captcha"
+                elif "turnstile" in sample:
+                    body_hint = "turnstile"
+                elif "loading" in sample:
+                    body_hint = "loading"
+            log.info(
+                "RESP ← %s status=%s type=%s len=%s server=%s location=%s cf-ray=%s hint=%s",
+                host or url[:60], status, content_type or "-", body_len,
+                server or "-", location or "-", cf_ray or "-", body_hint,
+            )
+
     async def start(self):
-        srv = await asyncio.start_server(self._on_client, self.host, self.port)
+        http_srv = await asyncio.start_server(self._on_client, self.host, self.port)
+        socks_srv = None
+
+        if self.socks_enabled:
+            try:
+                socks_srv = await asyncio.start_server(
+                    self._on_socks_client, self.socks_host, self.socks_port
+                )
+            except OSError as e:
+                log.error("SOCKS5 listener failed on %s:%d: %s",
+                          self.socks_host, self.socks_port, e)
+
         log.info(
-            "Listening on %s:%d — configure your browser HTTP proxy to this address",
+            "HTTP proxy listening on %s:%d",
             self.host, self.port,
         )
-        async with srv:
+        if socks_srv:
-            await srv.serve_forever()
+            log.info(
+                "SOCKS5 proxy listening on %s:%d",
+                self.socks_host, self.socks_port,
+            )
+
+        async with http_srv:
+            if socks_srv:
+                async with socks_srv:
+                    await asyncio.gather(
+                        http_srv.serve_forever(),
+                        socks_srv.serve_forever(),
+                    )
+            else:
+                await http_srv.serve_forever()
 
     # ── client handler ────────────────────────────────────────────
 
@@ -176,6 +319,69 @@ class ProxyServer:
             except Exception:
                 pass
 
+    async def _on_socks_client(self, reader: asyncio.StreamReader,
+                               writer: asyncio.StreamWriter):
+        addr = writer.get_extra_info("peername")
+        try:
+            header = await asyncio.wait_for(reader.readexactly(2), timeout=15)
+            ver, nmethods = header[0], header[1]
+            if ver != 5:
+                return
+
+            methods = await asyncio.wait_for(reader.readexactly(nmethods), timeout=10)
+            if 0x00 not in methods:
+                writer.write(b"\x05\xff")
+                await writer.drain()
+                return
+
+            writer.write(b"\x05\x00")
+            await writer.drain()
+
+            req = await asyncio.wait_for(reader.readexactly(4), timeout=15)
+            ver, cmd, _rsv, atyp = req
+            if ver != 5 or cmd != 0x01:
+                writer.write(b"\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00")
+                await writer.drain()
+                return
+
+            if atyp == 0x01:
+                raw = await asyncio.wait_for(reader.readexactly(4), timeout=10)
+                host = socket.inet_ntoa(raw)
+            elif atyp == 0x03:
+                ln = (await asyncio.wait_for(reader.readexactly(1), timeout=10))[0]
+                host = (await asyncio.wait_for(reader.readexactly(ln), timeout=10)).decode(
+                    errors="replace"
+                )
+            elif atyp == 0x04:
+                raw = await asyncio.wait_for(reader.readexactly(16), timeout=10)
+                host = socket.inet_ntop(socket.AF_INET6, raw)
+            else:
+                writer.write(b"\x05\x08\x00\x01\x00\x00\x00\x00\x00\x00")
+                await writer.drain()
+                return
+
+            port_raw = await asyncio.wait_for(reader.readexactly(2), timeout=10)
+            port = int.from_bytes(port_raw, "big")
+
+            log.info("SOCKS5 CONNECT → %s:%d", host, port)
+
+            writer.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
+            await writer.drain()
+            await self._handle_target_tunnel(host, port, reader, writer)
+
+        except asyncio.IncompleteReadError:
+            pass
+        except asyncio.TimeoutError:
+            log.debug("SOCKS5 timeout: %s", addr)
+        except Exception as e:
+            log.error("SOCKS5 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):
@@ -189,6 +395,13 @@ class ProxyServer:
         writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
         await writer.drain()
 
+        await self._handle_target_tunnel(host, port, reader, writer)
+
+    async def _handle_target_tunnel(self, host: str, port: int,
+                                    reader: asyncio.StreamReader,
+                                    writer: asyncio.StreamWriter):
+        """Route a target connection through the active relay mode."""
+
         if self.mode == "apps_script":
             override_ip = self._sni_rewrite_ip(host)
             if override_ip:
@@ -200,10 +413,29 @@ class ProxyServer:
                 await self._do_sni_rewrite_tunnel(host, port, reader, writer,
                                                   connect_ip=override_ip)
             elif self._is_google_domain(host):
+                if self._direct_temporarily_disabled(host):
+                    log.info("Relay fallback → %s (direct tunnel temporarily disabled)", host)
+                    if port == 443:
+                        await self._do_mitm_connect(host, port, reader, writer)
+                    else:
+                        await self._do_plain_http_tunnel(host, port, reader, writer)
+                    return
+
                 log.info("Direct tunnel → %s (Google domain, skipping relay)", host)
-                await self._do_direct_tunnel(host, port, reader, writer)
+                ok = await self._do_direct_tunnel(host, port, reader, writer)
-            else:
+                if ok:
+                    return
+
+                self._remember_direct_failure(host)
+                log.warning("Direct tunnel fallback → %s (switching to relay)", host)
+                if port == 443:
+                    await self._do_mitm_connect(host, port, reader, writer)
+                else:
+                    await self._do_plain_http_tunnel(host, port, reader, writer)
+            elif port == 443:
                 await self._do_mitm_connect(host, port, reader, writer)
+            else:
+                await self._do_plain_http_tunnel(host, port, reader, writer)
         else:
             await self.fronter.tunnel(host, port, reader, writer)
 
@@ -271,25 +503,132 @@ class ProxyServer:
     # 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_OWNED_SUFFIXES = (
         ".google.com", ".google.co",
         ".googleapis.com", ".gstatic.com",
         ".googleusercontent.com",
     )
-    _GOOGLE_EXACT = {
+    _GOOGLE_OWNED_EXACT = {
         "google.com", "gstatic.com", "googleapis.com",
     }
 
     def _is_google_domain(self, host: str) -> bool:
-        """Return True if host is a Google-owned domain."""
+        """Return True if host should use the raw direct Google shortcut."""
         h = host.lower().rstrip(".")
-        if h in self._GOOGLE_EXACT:
+        if self._is_direct_google_excluded(h):
+            return False
+        if not self._is_google_owned_domain(h):
+            return False
+        return self._is_direct_google_allowed(h)
+
+    def _is_google_owned_domain(self, host: str) -> bool:
+        if host in self._GOOGLE_OWNED_EXACT:
             return True
-        for suffix in self._GOOGLE_SUFFIXES:
+        for suffix in self._GOOGLE_OWNED_SUFFIXES:
-            if h.endswith(suffix):
+            if host.endswith(suffix):
                 return True
         return False
 
+    def _is_direct_google_excluded(self, host: str) -> bool:
+        if host in self._direct_google_exclude:
+            return True
+        for suffix in self._GOOGLE_DIRECT_SUFFIX_EXCLUDE:
+            if host.endswith(suffix):
+                return True
+        for token in self._direct_google_exclude:
+            if token.startswith(".") and host.endswith(token):
+                return True
+        return False
+
+    def _is_direct_google_allowed(self, host: str) -> bool:
+        if host in self._direct_google_allow:
+            return True
+        for suffix in self._GOOGLE_DIRECT_ALLOW_SUFFIXES:
+            if host.endswith(suffix):
+                return True
+        for token in self._direct_google_allow:
+            if token.startswith(".") and host.endswith(token):
+                return True
+        return False
+
+    def _direct_temporarily_disabled(self, host: str) -> bool:
+        h = host.lower().rstrip(".")
+        now = time.time()
+        disabled = False
+        for key in self._direct_failure_keys(h):
+            until = self._direct_fail_until.get(key, 0)
+            if until > now:
+                disabled = True
+            else:
+                self._direct_fail_until.pop(key, None)
+        return disabled
+
+    def _remember_direct_failure(self, host: str, ttl: int = 600):
+        until = time.time() + ttl
+        for key in self._direct_failure_keys(host.lower().rstrip(".")):
+            self._direct_fail_until[key] = until
+
+    def _direct_failure_keys(self, host: str) -> tuple[str, ...]:
+        keys = [host]
+        if host.endswith(".google.com") or host == "google.com":
+            keys.append("*.google.com")
+        if host.endswith(".googleapis.com") or host == "googleapis.com":
+            keys.append("*.googleapis.com")
+        if host.endswith(".gstatic.com") or host == "gstatic.com":
+            keys.append("*.gstatic.com")
+        if host.endswith(".googleusercontent.com") or host == "googleusercontent.com":
+            keys.append("*.googleusercontent.com")
+        return tuple(dict.fromkeys(keys))
+
+    async def _open_tcp_connection(self, target: str, port: int,
+                                   timeout: float = 10.0):
+        """Connect with IPv4-first resolution and clearer failure reporting."""
+        errors: list[str] = []
+        loop = asyncio.get_running_loop()
+
+        try:
+            ipaddress.ip_address(target)
+            candidates = [(0, target)]
+        except ValueError:
+            try:
+                infos = await asyncio.wait_for(
+                    loop.getaddrinfo(
+                        target,
+                        port,
+                        family=socket.AF_UNSPEC,
+                        type=socket.SOCK_STREAM,
+                    ),
+                    timeout=timeout,
+                )
+            except Exception as exc:
+                raise OSError(f"dns lookup failed for {target}: {exc!r}") from exc
+
+            candidates = []
+            seen = set()
+            for family, _type, _proto, _canon, sockaddr in infos:
+                ip = sockaddr[0]
+                key = (family, ip)
+                if key in seen:
+                    continue
+                seen.add(key)
+                candidates.append((family, ip))
+
+            candidates.sort(key=lambda item: 0 if item[0] == socket.AF_INET else 1)
+
+        for family, ip in candidates:
+            try:
+                return await asyncio.wait_for(
+                    asyncio.open_connection(ip, port, family=family or 0),
+                    timeout=timeout,
+                )
+            except Exception as exc:
+                fam = "ipv4" if family == socket.AF_INET else (
+                    "ipv6" if family == socket.AF_INET6 else "auto"
+                )
+                errors.append(f"{ip} ({fam}): {exc!r}")
+
+        raise OSError("; ".join(errors) or f"connect failed for {target}:{port}")
+
     # ── Direct tunnel (no MITM) ───────────────────────────────────
 
     async def _do_direct_tunnel(self, host: str, port: int,
@@ -300,17 +639,17 @@ class ProxyServer:
 
         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.
+        Without an override we connect to the real hostname so browser-safe
+        Google properties (Gemini assets, Play, Accounts, etc.) use their
+        normal edge instead of being forced onto the fronting IP.
         """
-        target_ip = connect_ip or self.fronter.connect_host
+        target_ip = connect_ip or host
         try:
-            r_remote, w_remote = await asyncio.wait_for(
+            r_remote, w_remote = await self._open_tcp_connection(target_ip, port, timeout=10)
-                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
+            return False
 
         async def pipe(src, dst, label):
             try:
@@ -334,6 +673,7 @@ class ProxyServer:
             pipe(reader, w_remote, f"client→{host}"),
             pipe(r_remote, writer, f"{host}→client"),
         )
+        return True
 
     # ── SNI-rewrite tunnel ────────────────────────────────────────
 
@@ -407,6 +747,11 @@ class ProxyServer:
 
     # ── MITM CONNECT (apps_script mode) ───────────────────────────
 
+    async def _do_plain_http_tunnel(self, host: str, port: int, reader, writer):
+        """Handle plain HTTP over SOCKS5 in apps_script mode."""
+        log.info("Plain HTTP relay → %s:%d", host, port)
+        await self._relay_http_stream(host, port, reader, writer)
+
     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)
@@ -433,6 +778,10 @@ class ProxyServer:
         # Update writer to use the new TLS transport
         writer._transport = new_transport
 
+        await self._relay_http_stream(host, port, reader, writer)
+
+    async def _relay_http_stream(self, host: str, port: int, reader, writer):
+        """Read decrypted/origin-form HTTP requests and relay them."""
         # Read and relay HTTP requests from the browser (now decrypted)
         while True:
             try:
@@ -471,11 +820,16 @@ class ProxyServer:
                         k, v = raw_line.decode(errors="replace").split(":", 1)
                         headers[k.strip()] = v.strip()
 
-                # Build full URL (browser sends just the path in CONNECT)
+                # MITM traffic arrives as origin-form paths; SOCKS/plain HTTP can
-                if port == 443:
+                # also send absolute-form requests. Normalize both to full URLs.
+                if path.startswith("http://") or path.startswith("https://"):
+                    url = path
+                elif port == 443:
                     url = f"https://{host}{path}"
+                elif port == 80:
+                    url = f"http://{host}{path}"
                 else:
-                    url = f"https://{host}:{port}{path}"
+                    url = f"http://{host}:{port}{path}"
 
                 log.info("MITM → %s %s", method, url)
 
@@ -502,7 +856,7 @@ class ProxyServer:
 
                 # Check local cache first (GET only)
                 response = None
-                if method == "GET" and not body:
+                if self._cache_allowed(method, url, headers, body):
                     response = self._cache.get(url)
                     if response:
                         log.debug("Cache HIT: %s", url[:60])
@@ -522,7 +876,7 @@ class ProxyServer:
                         )
 
                     # Cache successful GET responses
-                    if method == "GET" and not body and response:
+                    if self._cache_allowed(method, url, headers, body) and response:
                         ttl = ResponseCache.parse_ttl(response, url)
                         if ttl > 0:
                             self._cache.put(url, response, ttl)
@@ -533,6 +887,8 @@ class ProxyServer:
                 if origin and response:
                     response = self._inject_cors_headers(response, origin)
 
+                self._log_response_summary(url, response)
+
                 writer.write(response)
                 await writer.drain()
 
@@ -692,7 +1048,7 @@ class ProxyServer:
 
             # Cache check for GET
             response = None
-            if method == "GET" and not body:
+            if self._cache_allowed(method, url, headers, body):
                 response = self._cache.get(url)
                 if response:
                     log.debug("Cache HIT (HTTP): %s", url[:60])
@@ -700,7 +1056,7 @@ class ProxyServer:
             if response is None:
                 response = await self._relay_smart(method, url, headers, body)
                 # Cache successful GET
-                if method == "GET" and not body and response:
+                if self._cache_allowed(method, url, headers, body) and response:
                     ttl = ResponseCache.parse_ttl(response, url)
                     if ttl > 0:
                         self._cache.put(url, response, ttl)
@@ -708,6 +1064,7 @@ class ProxyServer:
             # Inject CORS headers for cross-origin requests
             if origin and response:
                 response = self._inject_cors_headers(response, origin)
+            self._log_response_summary(url, response)
         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)