Create app.py

app.py

@@ -0,0 +1,268 @@
+import time
+import re
+import socket
+import ipaddress
+from urllib.parse import urlparse
+
+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"),
+    ipaddress.ip_network("127.0.0.0/8"),
+    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("::1/128"),
+    ipaddress.ip_network("fc00::/7"),      # unique local
+    ipaddress.ip_network("fe80::/10"),     # link-local
+]
+
+DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))*\.?$")
+
+def _is_private_ip(ip_str: str) -> bool:
+    try:
+        ip = ipaddress.ip_address(ip_str)
+        return any(ip in net for net in PRIVATE_NETS)
+    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
+    """
+    t = (target or "").strip()
+    if not t:
+        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
+
+    # IP?
+    try:
+        ipaddress.ip_address(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(".")
+    if DOMAIN_RE.match(d):
+        return ("domain", d), d
+
+    return ("unknown", t), 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()
+            ans = dns.resolver.resolve(host, rtype, lifetime=3)
+            out[rtype] = [r.to_text() for r in ans]
+            out[f"{rtype}_ms"] = int((time.time() - start) * 1000)
+        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]
+    except Exception:
+        pass
+
+    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 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)
+        if code in (401, 403):
+            return f"REACHABLE but ACCESS_DENIED ({code})"
+        if code == 451:
+            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)"
+    if "tls" in errs.lower():
+        return "NOT_REACHABLE (TLS issue)"
+    return f"NOT_REACHABLE ({errs or 'unknown'})"
+
+def check_one(target: str):
+    (kind, _), host = _normalize_target(target)
+
+    if not target or not host:
+        return {"error": "Please enter a domain / IP / URL"}
+
+    # 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
+    try:
+        ipaddress.ip_address(host)
+        if _is_private_ip(host):
+            return {"error": "Blocked: private/reserved IP not allowed (SSRF protection)."}
+        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))
+
+    status = classify(dns_info, http_results)
+
+    return {
+        "input": target.strip(),
+        "host": host_for_http,
+        "dns": dns_info,
+        "http": http_results[:2],
+        "status": status,
+        "note": "Result is from Hugging Face Space network (egress)."
+    }
+
+def bulk_check(domain: str, subdomains_text: str):
+    base = (domain or "").strip().rstrip(".")
+    if not base:
+        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}")
+
+    rows = []
+    for t in targets:
+        r = check_one(t)
+        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 ""),
+        })
+    return rows
+
+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."
+    )
+
+    with gr.Tab("Single Check"):
+        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)
+
+    with gr.Tab("Subdomain List (Your List Only)"):
+        base = gr.Textbox(label="Base domain", placeholder="example.com")
+        subs = gr.Textbox(label="Subdomains (one per line)", lines=10, placeholder="www\napi\ncdn\nor full: api.example.com")
+        btn2 = gr.Button("Bulk Check")
+        table = gr.Dataframe(
+            headers=["target", "status", "A", "AAAA", "http_code"],
+            datatype=["str", "str", "str", "str", "str"],
+            row_count=5,
+            col_count=(5, "fixed"),
+            label="Results"
+        )
+        btn2.click(bulk_check, inputs=[base, subs], outputs=table)
+
+demo.launch()