117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
#!/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}
|