Wallet system (sats ledger + deposit + spend), HydraProxy fulfillment queue, paid IP-quality scorer (85/100 verified)

quality.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Proxy/IP quality scorer — battery of checks via free APIs + DNSBL.
+
+Scores 0-100: connectivity, residential/ISP type, proxy-flag, blacklists,
+latency, and header-leak anonymity. No API keys required.
+"""
+import json
+import socket
+import subprocess
+import time
+
+
+def _curl(proxy, url, timeout=20):
+    cmd = ["curl", "-s", "--max-time", str(timeout)]
+    if proxy:
+        cmd += ["-x", proxy]
+    cmd.append(url)
+    t0 = time.time()
+    try:
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 5)
+        out = (r.stdout or "").strip()
+    except subprocess.TimeoutExpired:
+        out = ""
+    return out, (time.time() - t0) * 1000
+
+
+def _dnsbl(ip):
+    rev = ".".join(reversed(ip.split(".")))
+    hits = []
+    for bl in ("zen.spamhaus.org", "bl.spamcop.net", "b.barracudacentral.org"):
+        try:
+            socket.gethostbyname(f"{rev}.{bl}")
+            hits.append(bl)
+        except socket.gaierror:
+            pass
+    return hits
+
+
+def score_proxy(proxy=None):
+    checks = []
+    score = 0
+
+    out, latency = _curl(proxy, "http://ip-api.com/json"
+                         "?fields=status,query,country,city,regionName,isp,org,as,hosting,proxy,mobile")
+    try:
+        d = json.loads(out)
+    except Exception:
+        d = {}
+
+    if d.get("status") != "success":
+        checks.append(("Connectivity", False, "proxy failed to connect"))
+        return {"score": 0, "checks": checks, "breakdown": {}, "charged": True}
+
+    ip = d.get("query")
+    score += 30
+    checks.append(("Connectivity", True, f"egress {ip}"))
+
+    hosting = bool(d.get("hosting"))
+    if not hosting:
+        score += 25
+        checks.append(("Residential / ISP", True, f"{d.get('isp')} ({d.get('org')})"))
+    else:
+        checks.append(("Residential / ISP", False, "datacenter / hosting IP"))
+
+    flagged = bool(d.get("proxy"))
+    if not flagged:
+        score += 20
+        checks.append(("Not proxy-flagged", True, "clean"))
+    else:
+        checks.append(("Not proxy-flagged", False, "IP flagged as proxy/VPN"))
+
+    bl = _dnsbl(ip)
+    if not bl:
+        score += 15
+        checks.append(("Blacklists", True, "no hits"))
+    else:
+        checks.append(("Blacklists", False, ", ".join(bl)))
+
+    if latency < 2000:
+        score += 10
+        checks.append(("Latency", True, f"{latency:.0f}ms"))
+    elif latency < 5000:
+        score += 5
+        checks.append(("Latency", True, f"{latency:.0f}ms (slow)"))
+    else:
+        checks.append(("Latency", False, f"{latency:.0f}ms"))
+
+    leak = []
+    try:
+        hout, _ = _curl(proxy, "https://httpbin.org/headers", timeout=15)
+        hd = json.loads(hout).get("headers", {})
+        for k in ("X-Forwarded-For", "Via", "X-Real-Ip"):
+            if k in hd:
+                leak.append(k)
+    except Exception:
+        pass
+
+    if leak:
+        checks.append(("Anonymity", False, f"leaks: {', '.join(leak)}"))
+    else:
+        checks.append(("Anonymity", True, "no header leaks"))
+
+    breakdown = {
+        "egress_ip": ip,
+        "geo": f"{d.get('city')}, {d.get('regionName')}, {d.get('country')}",
+        "isp": d.get("isp"),
+        "asn": d.get("as"),
+        "hosting": hosting,
+        "proxy_flagged": flagged,
+        "mobile": bool(d.get("mobile")),
+        "blacklists": bl,
+        "latency_ms": round(latency),
+        "leak_headers": leak,
+    }
+
+    return {"score": score, "checks": checks, "breakdown": breakdown, "charged": True}