🔒 Enforce TLS Certificate Validation across Proxy and Transports

src/domain_fronter.py

@@ -64,7 +64,6 @@ def __init__(self, config: dict):
         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]] = []
@@ -96,7 +95,7 @@ def __init__(self, config: dict):
                 from h2_transport import H2Transport, H2_AVAILABLE
                 if H2_AVAILABLE:
                     self._h2 = H2Transport(
-                        self.connect_host, self.sni_host, self.verify_ssl
+                        self.connect_host, self.sni_host
                     )
                     log.info("HTTP/2 multiplexing available — "
                              "all requests will share one connection")
@@ -107,9 +106,6 @@ def __init__(self, config: dict):
 
     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):

src/h2_transport.py

@@ -60,11 +60,9 @@ class H2Transport:
       - Configurable max concurrency
     """
 
-    def __init__(self, connect_host: str, sni_host: str,
-                 verify_ssl: bool = True):
+    def __init__(self, connect_host: str, sni_host: str):
         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
@@ -102,9 +100,6 @@ async def _do_connect(self):
         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

src/proxy_server.py

@@ -364,9 +364,6 @@ async def _do_sni_rewrite_tunnel(self, host: str, port: int, reader, writer,
 
         # 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(

test_ssl_validation.py

@@ -0,0 +1,47 @@
+import asyncio
+from src.domain_fronter import DomainFronter
+from src.h2_transport import H2Transport
+
+async def test():
+    config = {
+        "mode": "apps_script",
+        "script_id": "test_id",
+        "verify_ssl": False # Should no longer affect logic!
+    }
+
+    fronter = DomainFronter(config)
+    print("DomainFronter initialized successfully.")
+
+    if hasattr(fronter, "verify_ssl"):
+        print("FAIL: DomainFronter still has verify_ssl attribute.")
+        return False
+
+    try:
+        ctx = fronter._ssl_ctx()
+        print(f"DomainFronter SSL verify mode: {ctx.verify_mode}")
+    except Exception as e:
+        print(f"Error checking DomainFronter SSL Context: {e}")
+        return False
+
+    try:
+        h2 = H2Transport("127.0.0.1", "localhost")
+        print("H2Transport initialized successfully.")
+
+        if hasattr(h2, "verify_ssl"):
+            print("FAIL: H2Transport still has verify_ssl attribute.")
+            return False
+
+    except Exception as e:
+        print(f"Error initializing H2Transport: {e}")
+        return False
+
+    return True
+
+if __name__ == "__main__":
+    success = asyncio.run(test())
+    if success:
+        print("TEST PASSED")
+        exit(0)
+    else:
+        print("TEST FAILED")
+        exit(1)