diagnose HF Cloudflare transport boundary

app.py

@@ -469,3 +469,125 @@ async def list_witness_events(session_id: str, limit: int = Query(50, ge=1, le=2
 @app.get("/health")
 async def health():
     return {"status": "ok", "timestamp": _now()}
+
+
+# ── Temporary transport diagnostic (S4.5D) ────────────────────
+@app.get("/_diag/transport")
+async def transport_diagnostic():
+    """Read-only HF-runtime transport probe. No D1 query or secret values."""
+    import asyncio
+    import socket
+    import ssl
+    import time
+    import httpx
+
+    host = "api.cloudflare.com"
+    result = {
+        "target": host,
+        "proxy_env_present": {
+            key: key in os.environ
+            for key in (
+                "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY",
+                "http_proxy", "https_proxy", "all_proxy", "no_proxy",
+            )
+        },
+        "dns": [],
+        "tcp": [],
+        "tls": [],
+        "https": [],
+    }
+
+    def run_socket_probes():
+        started = time.monotonic()
+        infos = socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)
+        seen = set()
+        targets = []
+        for family, socktype, proto, _canon, sockaddr in infos:
+            key = (family, sockaddr[0])
+            if key in seen:
+                continue
+            seen.add(key)
+            targets.append((family, socktype, proto, sockaddr))
+            result["dns"].append({
+                "family": "IPv6" if family == socket.AF_INET6 else "IPv4",
+                "address": sockaddr[0],
+            })
+
+        selected = {}
+        for target in targets:
+            family = target[0]
+            selected.setdefault(family, target)
+
+        for family, socktype, proto, sockaddr in selected.values():
+            label = "IPv6" if family == socket.AF_INET6 else "IPv4"
+            t0 = time.monotonic()
+            sock = socket.socket(family, socktype, proto)
+            sock.settimeout(3.0)
+            try:
+                sock.connect(sockaddr)
+                result["tcp"].append({
+                    "family": label, "ok": True,
+                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                })
+            except Exception as exc:
+                result["tcp"].append({
+                    "family": label, "ok": False,
+                    "error": type(exc).__name__,
+                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                })
+            finally:
+                sock.close()
+
+            t0 = time.monotonic()
+            raw = socket.socket(family, socktype, proto)
+            raw.settimeout(3.0)
+            try:
+                raw.connect(sockaddr)
+                context = ssl.create_default_context()
+                with context.wrap_socket(raw, server_hostname=host) as tls_sock:
+                    result["tls"].append({
+                        "family": label, "ok": True,
+                        "version": tls_sock.version(),
+                        "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                    })
+            except Exception as exc:
+                result["tls"].append({
+                    "family": label, "ok": False,
+                    "error": type(exc).__name__,
+                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                })
+                raw.close()
+
+        result["dns_elapsed_ms"] = round((time.monotonic() - started) * 1000, 1)
+
+    await asyncio.to_thread(run_socket_probes)
+
+    urls = (
+        "https://api.cloudflare.com/client/v4/",
+        "https://github.com/",
+        "https://huggingface.co/",
+    )
+    for trust_env in (True, False):
+        timeout = httpx.Timeout(5.0, connect=5.0)
+        async with httpx.AsyncClient(
+            timeout=timeout, trust_env=trust_env, follow_redirects=False
+        ) as client:
+            for url in urls:
+                t0 = time.monotonic()
+                entry = {"url": url, "trust_env": trust_env}
+                try:
+                    response = await client.get(url)
+                    entry.update({
+                        "ok": True,
+                        "status": response.status_code,
+                        "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                    })
+                except Exception as exc:
+                    entry.update({
+                        "ok": False,
+                        "error": type(exc).__name__,
+                        "elapsed_ms": round((time.monotonic() - t0) * 1000, 1),
+                    })
+                result["https"].append(entry)
+
+    return result

test_transport_diag.py

@@ -0,0 +1,20 @@
+import unittest
+
+import app
+
+
+class TransportDiagnosticRouteTest(unittest.TestCase):
+    def test_read_only_transport_diagnostic_route_exists(self):
+        routes = {
+            (route.path, tuple(sorted(route.methods or [])))
+            for route in app.app.routes
+        }
+        self.assertIn(
+            ("/_diag/transport", ("GET",)),
+            routes,
+            "Expected a temporary GET-only transport diagnostic endpoint",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()