diff --git a/android-qualify.service b/android-qualify.service new file mode 100644 index 0000000..37a3b5c --- /dev/null +++ b/android-qualify.service @@ -0,0 +1,12 @@ +[Unit] +Description=Proxy IG/Google qualification loop (gates proxies before account flows) +After=network.target + +[Service] +Type=simple +ExecStart=/usr/bin/python3 /opt/android-fleet/proxy_qualify_loop.py +Restart=always +RestartSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/app.py b/app.py index f018544..57c1f1f 100644 --- a/app.py +++ b/app.py @@ -982,10 +982,13 @@ def api_text(mac): serial = _serial_for(mac) if not serial: return jsonify({"error": "offline"}), 404 - txt = (data.get("text") or "").replace(" ", "%s").replace("'", "\\'") - r = subprocess.run(["adb", "-s", serial, "shell", f"input text '{txt}'"], - capture_output=True, text=True, timeout=20) - return jsonify({"ok": r.returncode == 0, "err": r.stderr[:100]}) + txt = (data.get("text") or "").replace(" ", "%s") + try: + r = subprocess.run(["adb", "-s", serial, "shell", "input", "text", txt], + capture_output=True, text=True, timeout=30) + return jsonify({"ok": r.returncode == 0, "err": r.stderr[:100]}) + except Exception as e: + return jsonify({"ok": False, "err": f"text err: {type(e).__name__}"}), 500 @app.route("/api/device//launch", methods=["POST"]) def api_launch(mac): diff --git a/proxy_qualify.py b/proxy_qualify.py new file mode 100644 index 0000000..a3def31 --- /dev/null +++ b/proxy_qualify.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +""" +Proxy Qualifier — gate every proxy before it enters an account-creation flow. +Per candidate: alive+latency -> egress IP -> ip-api cleanliness -> Spamhaus DNSBL +-> ACTIVE Instagram probe (i.instagram.com) -> Google probe (accounts.google.com). +Only proxies passing ALL get 'ig_ok' and are eligible for the factory. +""" +import concurrent.futures as cf +import json +import re +import socket +import subprocess +import sys +import time +import urllib.request + +DB = "/opt/android-fleet/fleet.db" +PROBE_IG = "https://i.instagram.com/api/v1/web/search/topsearch/?query=instagram" +PROBE_GOOGLE = "https://accounts.google.com/" +UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"} + +def http_proxy(p): + proto, host, port = p["proto"], p["host"], p["port"] + return {proto: f"{proto}://{host}:{port}"} if proto == "http" else {"https": f"socks5h://{host}:{port}"} if proto == "socks5" else {"https": f"socks4://{host}:{port}"} + +def get(url, proxy, timeout=15): + req = urllib.request.Request(url, headers=UA) + op = urllib.request.build_opener(urllib.request.ProxyHandler(proxy)) + with op.open(req, timeout=timeout) as r: + return r.status, r.read(400).decode(errors="ignore") + +def egress(proxy): + try: + st, body = get("https://api.ipify.org?format=json", proxy) + return json.loads(body).get("ip") if st == 200 else None + except Exception: + try: + st, body = get("http://ip-api.com/json/?fields=query", proxy, 10) + return json.loads(body).get("query") if st == 200 else None + except Exception: + return None + +def ipapi(ip): + try: + st, body = get(f"http://ip-api.com/json/{ip}?fields=status,country,isp,hosting,proxy,mobile", {}, 10) + return json.loads(body) if st == 200 else {} + except Exception: + return {} + +def spamhaus(ip): + try: + rev = ".".join(reversed(ip.split("."))) + socket.gethostbyname(f"{rev}.zen.spamhaus.org") + return True # listed + except socket.gaierror: + return False + except Exception: + return None + +def probe_ig(proxy): + """Instagram verdict: any IG response = reachable; 429 = flagged; timeout = dead.""" + try: + st, body = get(PROBE_IG, proxy) + if st == 429: + return "rate_limited", st + # 200/401/404/405/etc all mean Instagram's servers ANSWERED this IP + return "reachable", st + except Exception as e: + return "unreachable", str(e)[:60] + +def probe_google(proxy): + try: + st, _ = get(PROBE_GOOGLE, proxy) + return ("pass" if st in (200, 301, 302) else f"http{st}"), st + except Exception as e: + return "unreachable", str(e)[:60] + +def qualify(p): + res = {"name": p["name"], "host": p["host"], "port": p["port"], "proto": p["proto"]} + px = http_proxy(p) + t0 = time.time() + ip = egress(px) + res["latency_ms"] = int((time.time() - t0) * 1000) + if not ip: + res["verdict"] = "dead" + return res + res["egress"] = ip + info = ipapi(ip) + res["hosting"] = bool(info.get("hosting")) + res["proxy_flag"] = bool(info.get("proxy")) + res["mobile"] = bool(info.get("mobile")) + res["country"] = info.get("country") + res["spamhaus"] = spamhaus(ip) + ig, igcode = probe_ig(px) + res["ig"] = ig + res["ig_code"] = igcode + g, gcode = probe_google(px) + res["google"] = g + res["google_code"] = gcode + clean = (not res["hosting"]) and (not res["proxy_flag"]) and (not res["spamhaus"]) + res["clean"] = clean + # transparent proxies leak the LOCAL egress — they don't actually tunnel. + local = res["egress"] in ("76.146.9.44", "10.30.20.85") + res["transparent_leak"] = local + ig_ok = (ig == "reachable" or ig == "rate_limited") and not local + res["verdict"] = "ig_ok" if (ig_ok and g == "pass" and clean) else ( + "ig_only" if ig_ok else "fail") + return res + +def bank_proxies(): + import sqlite3 + con = sqlite3.connect(DB) + con.row_factory = sqlite3.Row + rows = con.execute("SELECT * FROM proxies WHERE health='healthy' ORDER BY clean DESC, latency_ms ASC LIMIT 40").fetchall() + con.close() + return [dict(r) for r in rows] + +if __name__ == "__main__": + mode = sys.argv[1] if len(sys.argv) > 1 else "bank" + if mode == "bank": + cands = bank_proxies() + else: # arbitrary list: "host:port:socks5" ... + cands = [] + for spec in sys.argv[1:]: + h, port, proto = spec.split(":") + cands.append({"name": f"manual-{h}", "host": h, "port": int(port), "proto": proto}) + print(f"qualifying {len(cands)} candidates...", flush=True) + results = [] + with cf.ThreadPoolExecutor(max_workers=8) as ex: + for r in ex.map(qualify, cands): + results.append(r) + for r in results: + print(json.dumps(r), flush=True) + ok = [r for r in results if r["verdict"] in ("ig_ok", "ig_only")] + print(f"\n=== {len(ok)}/{len(results)} proxies eligible for account flows ===", flush=True) + for r in ok: + print(f" {r['name']} {r['host']}:{r['port']} egress={r['egress']} country={r['country']} " + f"clean={r['clean']} ig={r['ig']} google={r['google']} lat={r['latency_ms']}ms", flush=True) diff --git a/proxy_qualify_loop.py b/proxy_qualify_loop.py new file mode 100644 index 0000000..e35a5aa --- /dev/null +++ b/proxy_qualify_loop.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""Qualify loop: runs the IG/Google gate on every unqualified healthy proxy in the +bank, tagging ig_status. Eligible proxies are the only ones assigned to devices.""" +import json +import sqlite3 +import subprocess +import time + +DB = "/opt/android-fleet/fleet.db" + +def main(): + con = sqlite3.connect(DB) + con.row_factory = sqlite3.Row + # ensure column + cols = [r[1] for r in con.execute("PRAGMA table_info(proxies)").fetchall()] + if "ig_status" not in cols: + con.execute("ALTER TABLE proxies ADD COLUMN ig_status TEXT DEFAULT 'unqualified'") + con.commit() + rows = con.execute("SELECT * FROM proxies WHERE health='healthy' AND ig_status='unqualified'").fetchall() + con.close() + if not rows: + return + # qualify via the CLI (bank mode would re-check all; we drive one by one) + for r in rows: + spec = f"{r['host']}:{r['port']}:{r['proto']}" + try: + out = subprocess.run(["python3", "/opt/android-fleet/proxy_qualify.py", spec], + capture_output=True, text=True, timeout=90).stdout + for line in out.splitlines(): + if line.startswith("{"): + d = json.loads(line) + con = sqlite3.connect(DB) + con.execute("UPDATE proxies SET ig_status=? WHERE host=? AND port=?", + (d["verdict"], r["host"], r["port"])) + con.commit() + con.close() + print(f"{spec}: {d['verdict']} (ig={d.get('ig')}, google={d.get('google')})", flush=True) + except Exception as e: + print(f"{spec}: qualify err {e}", flush=True) + +if __name__ == "__main__": + while True: + try: + main() + except Exception as e: + print("qualify loop err:", e, flush=True) + time.sleep(600)