#!/usr/bin/env python3 """ ProxyFly integration: harvest free proxies -> live-test -> cleanliness-grade (hosting/proxy flags + Spamhaus DNSBL) -> feed the fleet proxy bank. Every machine gets a verified egress IP, 100% backend, zero-touch. """ import json import os import socket import sqlite3 import threading import time import urllib.request from concurrent.futures import ThreadPoolExecutor DB_PATH = "/opt/android-fleet/fleet.db" SOURCES = [ "https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/all/data.json", ] MAX_FETCH = 200 # candidates per cycle MAX_TEST = 120 # live-test at most this many per cycle MAX_POOL = 40 # cap the proxies table (perf) TEST_TIMEOUT = 7 HARVEST_INTERVAL = 1500 # 25 min def log(msg): print(f"[harvest] {msg}", flush=True) def dbcon(): con = sqlite3.connect(DB_PATH, timeout=30) con.row_factory = sqlite3.Row return con def fetch_candidates(): cands = [] for src in SOURCES: try: with urllib.request.urlopen(src, timeout=45) as r: data = json.loads(r.read()) for p in data: proto = p.get("protocol", "") if proto not in ("socks5", "https", "http", "socks4"): continue if p.get("anonymity") == "transparent": continue # leaks real IP — useless for the fleet cands.append({ "host": p["ip"], "port": p["port"], "proto": proto, "country": (p.get("geolocation") or {}).get("country", "?"), }) except Exception as e: log(f"source error {src}: {e}") # dedupe by host:port seen = set() out = [] for c in cands: k = f"{c['host']}:{c['port']}" if k not in seen: seen.add(k) out.append(c) return out[:MAX_FETCH] def test_proxy(c): """Returns (c, egress_ip, latency_ms) or (c, None, None).""" proto_map = {"socks5": "socks5h://", "https": "https://", "http": "http://", "socks4": "socks4://"} url = proto_map[c["proto"]] + f"{c['host']}:{c['port']}" import urllib.request as ur opener = ur.build_opener(ur.ProxyHandler({"http": url, "https": url})) t0 = time.time() try: with opener.open("https://api.ipify.org", timeout=TEST_TIMEOUT) as r: ip = r.read().decode().strip() lat = int((time.time() - t0) * 1000) return (c, ip, lat) except Exception: return (c, None, None) def check_clean(ips): """ip-api batch: flags for each ip. Returns {ip: dict}.""" out = {} ips = [i for i in ips if i] for i in range(0, len(ips), 15): batch = ips[i:i + 15] try: req = urllib.request.Request( "http://ip-api.com/batch?fields=status,country,isp,as,hosting,proxy,mobile,query", data=json.dumps(batch).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=20) as r: res = json.loads(r.read()) for e in res: if e.get("status") == "success": out[e["query"]] = e except Exception as e: log(f"ip-api batch error: {e}") time.sleep(4) # 45 req/min limit — throttle hard return out def dnsbl_listed(ip): """Spamhaus ZEN: True if listed.""" try: rev = ".".join(reversed(ip.split("."))) + ".zen.spamhaus.org" socket.gethostbyname(rev) return True except socket.gaierror: return False except Exception: return False def prune_pool(): con = dbcon() rows = con.execute("SELECT id, egress_ip, source FROM proxies ORDER BY id").fetchall() # remove duplicates by egress ip (keep lowest id) seen = set() for r in rows: if r["egress_ip"] and r["egress_ip"] in seen: con.execute("DELETE FROM proxies WHERE id=?", (r["id"],)) elif r["egress_ip"]: seen.add(r["egress_ip"]) # cap pool size (drop oldest proxifly entries beyond cap) over = [r for r in rows if r["source"] == "proxifly"] if len(over) > MAX_POOL: for r in over[:-MAX_POOL]: con.execute("DELETE FROM proxies WHERE id=?", (r["id"],)) con.commit() con.close() def harvest(): log(f"cycle start — fetching {len(SOURCES)} source(s)") cands = fetch_candidates() log(f"{len(cands)} candidates, testing up to {MAX_TEST}") tested = [] with ThreadPoolExecutor(max_workers=20) as ex: for c, ip, lat in ex.map(test_proxy, cands[:MAX_TEST]): if ip: c["egress_ip"] = ip c["latency_ms"] = lat tested.append(c) log(f"{len(tested)} proxies alive; grading cleanliness") flags = check_clean([c["egress_ip"] for c in tested]) dns_flags = {} with ThreadPoolExecutor(max_workers=20) as ex: for c in tested: dns_flags[c["egress_ip"]] = ex.submit(dnsbl_listed, c["egress_ip"]) for k, f in dns_flags.items(): dns_flags[k] = f.result() con = dbcon() added = 0 for c in tested: f = flags.get(c["egress_ip"], {}) dns = dns_flags.get(c["egress_ip"], False) clean = 1 if (not f.get("hosting") and not f.get("proxy") and not dns) else 0 mobile = 1 if f.get("mobile") else 0 # keep only clean ones + a few fallbacks, skip dirty datacenter spam if clean == 0: why = "hosting" if f.get("hosting") else ("proxy" if f.get("proxy") else ("dnsbl" if dns else "unknown")) log(f"rejected {c['egress_ip']} ({why})") continue name = f"proxifly-{c['country']}-{c['egress_ip'].split('.')[-2]}" con.execute("""INSERT INTO proxies(name, host, port, proto, health, latency_ms, assigned_mac, last_check, clean, country, egress_ip, source) VALUES(?,?,?,?,'healthy',?,NULL,?,?,?,?,?) ON CONFLICT(egress_ip) DO UPDATE SET health='healthy', latency_ms=?, last_check=?, clean=?""", (name, c["host"], c["port"], c["proto"], c["latency_ms"], time.time(), clean, c["country"], c["egress_ip"], "proxifly", c["latency_ms"], time.time(), clean)) added += 1 con.commit() con.close() prune_pool() log(f"cycle done: +{added} clean proxies into the bank") assign_best() log("auto-assignment sweep complete") def assign_best(): """Give every online device without a proxy the best clean one available.""" con = dbcon() con.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_prox_egress ON proxies(egress_ip)") devs = con.execute("SELECT * FROM devices WHERE adb_state='online' AND ip IS NOT NULL").fetchall() for d in devs: have = con.execute("SELECT * FROM proxies WHERE assigned_mac=?", (d["mac"],)).fetchone() if have: continue # already assigned — keep stable best = con.execute("""SELECT * FROM proxies WHERE clean=1 AND health='healthy' AND assigned_mac IS NULL ORDER BY latency_ms ASC LIMIT 1""").fetchone() if not best: best = con.execute("""SELECT * FROM proxies WHERE health='healthy' AND assigned_mac IS NULL ORDER BY latency_ms ASC LIMIT 1""").fetchone() if not best: continue con.execute("UPDATE proxies SET assigned_mac=? WHERE id=?", (d["mac"], best["id"])) con.execute("UPDATE devices SET proxy_name=?, public_ip=? WHERE mac=?", (best["name"], best["egress_ip"], d["mac"])) log(f"assigned {best['name']} (egress {best['egress_ip']}, clean={best['clean']}) -> {d['mac'][:14]}") con.commit() con.close() def loop(): while True: time.sleep(HARVEST_INTERVAL) try: harvest() except Exception as e: log(f"cycle error: {e}") if __name__ == "__main__": log("proxyfly harvest engine starting") try: harvest() except Exception as e: log(f"startup harvest error: {e}") loop()