Update app.py

app.py

@@ -8,9 +8,6 @@ import httpx
 import dns.resolver
 import gradio as gr
 
-# -----------------------
-# Safety: SSRF protection
-# -----------------------
 PRIVATE_NETS = [
     ipaddress.ip_network("0.0.0.0/8"),
     ipaddress.ip_network("10.0.0.0/8"),
@@ -18,14 +15,21 @@ PRIVATE_NETS = [
     ipaddress.ip_network("169.254.0.0/16"),
     ipaddress.ip_network("172.16.0.0/12"),
     ipaddress.ip_network("192.168.0.0/16"),
-    ipaddress.ip_network("224.0.0.0/4"),   # multicast
-    ipaddress.ip_network("240.0.0.0/4"),   # reserved
+    ipaddress.ip_network("224.0.0.0/4"),
+    ipaddress.ip_network("240.0.0.0/4"),
     ipaddress.ip_network("::1/128"),
-    ipaddress.ip_network("fc00::/7"),      # unique local
-    ipaddress.ip_network("fe80::/10"),     # link-local
+    ipaddress.ip_network("fc00::/7"),
+    ipaddress.ip_network("fe80::/10"),
 ]
 
-DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))*\.?$")
+DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))*$")
+
+# ✅ Reuse one client (less event-loop weirdness + faster)
+CLIENT = httpx.Client(
+    follow_redirects=True,
+    timeout=8.0,
+    headers={"User-Agent": "HF-Connectivity-Checker/1.0"},
+)
 
 def _is_private_ip(ip_str: str) -> bool:
     try:
@@ -34,46 +38,32 @@ def _is_private_ip(ip_str: str) -> bool:
     except Exception:
         return True
 
-def _normalize_target(target: str) -> tuple[str, str]:
-    """
-    Returns (kind, value):
-      kind: "domain" | "ip" | "url"
-      value: normalized host (domain/ip) or full url
-    """
+def _parse_target(target: str):
     t = (target or "").strip()
     if not t:
-        return ("", "")
+        return ("", "", "")
 
-    # If user passed full URL
     if t.startswith("http://") or t.startswith("https://"):
         u = urlparse(t)
-        host = u.hostname or ""
-        return ("url", t), host
+        return ("url", t, u.hostname or "")
 
-    # IP?
     try:
         ipaddress.ip_address(t)
-        return ("ip", t), t
+        return ("ip", t, t)
     except Exception:
         pass
 
-    # Domain?
-    # Allow IDN via idna encoding by socket/dns; here just basic sanity
-    d = t.strip().rstrip(".")
+    d = t.rstrip(".")
     if DOMAIN_RE.match(d):
-        return ("domain", d), d
+        return ("domain", d, d)
 
-    return ("unknown", t), t
+    return ("unknown", t, "")
 
 def resolve_dns(host: str):
-    """
-    Return DNS data: A/AAAA/CNAME if possible.
-    """
     out = {"A": [], "AAAA": [], "CNAME": []}
     if not host:
         return out
 
-    # A/AAAA
     for rtype in ["A", "AAAA"]:
         try:
             start = time.time()
@@ -83,7 +73,6 @@ def resolve_dns(host: str):
         except Exception as e:
             out[f"{rtype}_error"] = str(e)
 
-    # CNAME
     try:
         ans = dns.resolver.resolve(host, "CNAME", lifetime=3)
         out["CNAME"] = [r.target.to_text().rstrip(".") for r in ans]
@@ -92,56 +81,39 @@ def resolve_dns(host: str):
 
     return out
 
-def check_http(url: str, timeout_s: float = 8.0):
-    """
-    Try HTTPS then HTTP. Return details.
-    """
-    results = []
-
-    def _try(one_url: str):
-        info = {"url": one_url}
-        try:
-            start = time.time()
-            with httpx.Client(follow_redirects=True, timeout=timeout_s, headers={"User-Agent": "HF-Connectivity-Checker/1.0"}) as client:
-                r = client.head(one_url)
-                # Some servers don't like HEAD; fallback to GET small
-                if r.status_code in (405, 403) or r.status_code >= 500:
-                    r = client.get(one_url, headers={"Range": "bytes=0-1024"})
-                elapsed = int((time.time() - start) * 1000)
-
-            info.update({
-                "ok": True,
-                "status_code": r.status_code,
-                "final_url": str(r.url),
-                "latency_ms": elapsed,
-                "server": r.headers.get("server", ""),
-                "via": r.headers.get("via", ""),
-                "cf_ray": r.headers.get("cf-ray", ""),
-            })
-        except httpx.ConnectTimeout:
-            info.update({"ok": False, "error": "connect_timeout"})
-        except httpx.ReadTimeout:
-            info.update({"ok": False, "error": "read_timeout"})
-        except httpx.ConnectError as e:
-            info.update({"ok": False, "error": f"connect_error: {e}"})
-        except httpx.HTTPError as e:
-            info.update({"ok": False, "error": f"http_error: {e}"})
-        except Exception as e:
-            info.update({"ok": False, "error": f"unknown_error: {e}"})
-        return info
-
-    results.append(_try(url))
-    return results
+def try_http(url: str):
+    info = {"url": url}
+    try:
+        start = time.time()
+        r = CLIENT.head(url)
+        if r.status_code in (405, 403) or r.status_code >= 500:
+            r = CLIENT.get(url, headers={"Range": "bytes=0-1024"})
+        elapsed = int((time.time() - start) * 1000)
+        info.update({
+            "ok": True,
+            "status_code": r.status_code,
+            "final_url": str(r.url),
+            "latency_ms": elapsed,
+            "server": r.headers.get("server", ""),
+            "via": r.headers.get("via", ""),
+            "cf_ray": r.headers.get("cf-ray", ""),
+        })
+    except httpx.ConnectTimeout:
+        info.update({"ok": False, "error": "connect_timeout"})
+    except httpx.ReadTimeout:
+        info.update({"ok": False, "error": "read_timeout"})
+    except httpx.ConnectError as e:
+        info.update({"ok": False, "error": f"connect_error: {e}"})
+    except httpx.HTTPError as e:
+        info.update({"ok": False, "error": f"http_error: {e}"})
+    except Exception as e:
+        info.update({"ok": False, "error": f"unknown_error: {e}"})
+    return info
 
 def classify(dns_info, http_info):
-    """
-    Simple classification from HF viewpoint.
-    """
-    # DNS failures
     if dns_info.get("A_error") and dns_info.get("AAAA_error") and not dns_info.get("CNAME"):
         return "DNS_FAIL (HF can't resolve)"
 
-    # HTTP checks
     ok = [x for x in http_info if x.get("ok")]
     if ok:
         code = ok[0].get("status_code", 0)
@@ -151,7 +123,6 @@ def classify(dns_info, http_info):
             return "REACHABLE but LEGAL_RESTRICTION (451)"
         return f"REACHABLE ({code})"
 
-    # Common block-ish signals
     errs = " | ".join(x.get("error", "") for x in http_info if x.get("error"))
     if "timeout" in errs:
         return "NOT_REACHABLE (timeout / possible block)"
@@ -160,56 +131,44 @@ def classify(dns_info, http_info):
     return f"NOT_REACHABLE ({errs or 'unknown'})"
 
 def check_one(target: str):
-    (kind, _), host = _normalize_target(target)
+    kind, raw, host = _parse_target(target)
 
-    if not target or not host:
-        return {"error": "Please enter a domain / IP / URL"}
+    if kind == "":
+        return {"error": "Enter a domain / IP / URL"}
+    if kind == "unknown" or not host:
+        return {"error": "Invalid input"}
 
-    # For URL, pull host and rebuild with scheme attempts
-    if kind == "unknown":
-        return {"error": "Invalid input. Enter domain, IP, or full URL."}
-
-    # If host is IP: block private/reserved
+    # IP safety
     try:
         ipaddress.ip_address(host)
         if _is_private_ip(host):
-            return {"error": "Blocked: private/reserved IP not allowed (SSRF protection)."}
+            return {"error": "Blocked: private/reserved IP not allowed."}
         dns_info = {"A": [host], "AAAA": [], "CNAME": []}
         host_for_http = host
     except Exception:
-        # Domain: resolve DNS
         dns_info = resolve_dns(host)
-        # If resolved IPs contain private -> block (SSRF protection)
         ips = (dns_info.get("A") or []) + (dns_info.get("AAAA") or [])
         for ip in ips:
             if _is_private_ip(ip):
                 return {"error": "Blocked: resolves to private/reserved IP (SSRF protection)."}
         host_for_http = host
 
-    # Build URLs to try
     urls = []
     if kind == "url":
-        # Try exactly as given first
-        urls.append(target.strip())
-        # Also try forcing https/http with host only
-        urls.append(f"https://{host_for_http}")
-        urls.append(f"http://{host_for_http}")
-    else:
-        urls = [f"https://{host_for_http}", f"http://{host_for_http}"]
-
-    http_results = []
-    for u in urls[:2]:  # keep tight: only 2 tries
-        http_results.extend(check_http(u))
+        urls.append(raw)
+    urls.append(f"https://{host_for_http}")
+    urls.append(f"http://{host_for_http}")
 
+    http_results = [try_http(urls[0]), try_http(urls[1])]  # only two tries
     status = classify(dns_info, http_results)
 
     return {
         "input": target.strip(),
         "host": host_for_http,
         "dns": dns_info,
-        "http": http_results[:2],
+        "http": http_results,
         "status": status,
-        "note": "Result is from Hugging Face Space network (egress)."
+        "note": "Checked from Hugging Face Space network."
     }
 
 def bulk_check(domain: str, subdomains_text: str):
@@ -218,23 +177,21 @@ def bulk_check(domain: str, subdomains_text: str):
         return []
 
     lines = [x.strip() for x in (subdomains_text or "").splitlines() if x.strip()]
-    # Allow either full subdomain or just prefix
     targets = []
-    for s in lines[:200]:  # limit
-        if "." in s:
-            targets.append(s)
-        else:
-            targets.append(f"{s}.{base}")
+    for s in lines[:200]:
+        targets.append(s if "." in s else f"{s}.{base}")
 
     rows = []
     for t in targets:
         r = check_one(t)
+        dns = r.get("dns", {}) if isinstance(r, dict) else {}
+        http = r.get("http", [{}]) if isinstance(r, dict) else [{}]
         rows.append({
             "target": t,
             "status": r.get("status") or r.get("error", "error"),
-            "A": ",".join((r.get("dns", {}) or {}).get("A", [])) if isinstance(r, dict) else "",
-            "AAAA": ",".join((r.get("dns", {}) or {}).get("AAAA", [])) if isinstance(r, dict) else "",
-            "http_code": (r.get("http", [{}])[0].get("status_code") if r.get("http") else ""),
+            "A": ",".join(dns.get("A", [])),
+            "AAAA": ",".join(dns.get("AAAA", [])),
+            "http_code": http[0].get("status_code", ""),
         })
     return rows
 
@@ -242,12 +199,12 @@ with gr.Blocks(title="HF Domain/IP Connectivity Checker") as demo:
     gr.Markdown(
         "## HF Domain/IP Connectivity Checker\n"
         "Checks DNS + HTTP reachability **from this Hugging Face Space**.\n"
-        "- No brute-force scanning. Subdomains must be **user-provided list**.\n"
-        "- Private/reserved IPs blocked for safety."
+        "- Subdomains must be **your provided list** (no brute-force scanning).\n"
+        "- Private/reserved IPs blocked (SSRF protection)."
     )
 
     with gr.Tab("Single Check"):
-        inp = gr.Textbox(label="Domain / IP / URL", placeholder="example.com  OR  https://example.com  OR  1.2.3.4")
+        inp = gr.Textbox(label="Domain / IP / URL", placeholder="example.com OR https://example.com OR 1.2.3.4")
         btn = gr.Button("Check")
         out = gr.JSON(label="Result")
         btn.click(check_one, inputs=inp, outputs=out)
@@ -260,9 +217,10 @@ with gr.Blocks(title="HF Domain/IP Connectivity Checker") as demo:
             headers=["target", "status", "A", "AAAA", "http_code"],
             datatype=["str", "str", "str", "str", "str"],
             row_count=5,
-            col_count=(5, "fixed"),
-            label="Results"
+            column_count=(5, "fixed"),  # ✅ new param
+            label="Results",
         )
         btn2.click(bulk_check, inputs=[base, subs], outputs=table)
 
-demo.launch()
+# ✅ SSR off to avoid Python 3.13 loop cleanup errors
+demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)