#!/usr/bin/env python3 """ Reddit account factory — one verified Reddit profile per fleet device. Browser-based (Chrome on device, rides the device proxy) + email verification via Gmail plus-addressing and the IMAP link-clicker. uiautomator CAN see Chrome's DOM (unlike Google's auth webview). """ import imaplib import json import os import random import re import sqlite3 import subprocess import time import urllib.request import urllib.parse BASE = "http://10.30.20.31:8080" TOKEN = "godhead-fleet-2026" GMAIL = "jones.henry666q@gmail.com" APP_PASS = "hkdbecbdkncmvshg" DB = "/opt/android-fleet/fleet.db" COUNTER = "/opt/android-fleet/rdt_counter" def api(path, method="GET", payload=None, timeout=120): data = json.dumps(payload).encode() if payload is not None else None req = urllib.request.Request(BASE + path, data=data, method=method, headers={"Content-Type": "application/json", "X-Auth-Token": TOKEN}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def shell(mac, cmd, timeout=60): try: return api(f"/api/device/{mac}/exec", "POST", {"cmd": cmd}, timeout).get("output", "") except Exception as e: return f"ERR {e}" def tap(mac, x, y): api(f"/api/device/{mac}/tap", "POST", {"x": int(x), "y": int(y)}, 30) def text(mac, s): api(f"/api/device/{mac}/text", "POST", {"text": s}, 40) def next_counter(): n = int(open(COUNTER).read()) if os.path.exists(COUNTER) else 0 n += 1 open(COUNTER, "w").write(str(n)) return n def dump(mac): shell(mac, "uiautomator dump /sdcard/ui.xml >/dev/null 2>&1; cat /sdcard/ui.xml", 30) try: req = urllib.request.Request(BASE + f"/api/device/{mac}/exec", data=json.dumps({"cmd": "cat /sdcard/ui.xml"}).encode(), headers={"Content-Type": "application/json", "X-Auth-Token": TOKEN}, method="POST") with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()).get("output", "") except Exception: return "" def find_in_dump(xml, *needles, kind="text"): """Find node containing any needle in text/content-desc/hint. Return center.""" pat = re.compile(r']*?(?:text|content-desc)="([^"]*)"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"') hits = [] for m in pat.finditer(xml): label = m.group(1) if any(n.lower() in label.lower() for n in needles): x1, y1, x2, y2 = (int(m.group(i)) for i in (2, 3, 4, 5)) hits.append(((x1 + x2) // 2, (y1 + y2) // 2, label)) return hits def wait_dump(mac, tries=6): for _ in range(tries): xml = dump(mac) if xml.strip().startswith("<"): return xml time.sleep(3) return "" REDDIT_URL = "https://old.reddit.com/register" # server-rendered, works on old engines def create_reddit(mac): n = next_counter() username = f"jones_{random.choice(['pulse','drift','echo','flare','ember','wisp','forge','zephyr'])}{n}" password = "J0nes123!xYz" email = f"rdt{n}+{GMAIL}" log = [] # open register page in Chrome (device proxy applies) shell(mac, f"am start -a android.intent.action.VIEW -d '{REDDIT_URL}' com.android.chrome") time.sleep(12) xml = wait_dump(mac) # Chrome first-run welcome screen? accept & continue, then reload the URL if "accept" in xml.lower(): hits = find_in_dump(xml, "Accept & continue", "Accept and continue") if hits: tap(mac, *hits[0][:2]); time.sleep(5) shell(mac, f"am start -a android.intent.action.VIEW -d '{REDDIT_URL}' com.android.chrome") time.sleep(12) xml = wait_dump(mac) if not xml: return {"status": "chrome_dump_failed", "log": log} # username field hits = find_in_dump(xml, "username") if not hits: hits = find_in_dump(xml, "choose a username") if not hits: return {"status": "username_field_missing", "log": log} tap(mac, *hits[0][:2]); time.sleep(2); text(mac, username); time.sleep(1) log.append(f"username: {username}") # password field (often next after username; re-dump) shell(mac, "input keyevent 4"); time.sleep(1) xml = wait_dump(mac) hits = find_in_dump(xml, "password") if not hits: return {"status": "password_field_missing", "log": log} tap(mac, *hits[0][:2]); time.sleep(2); text(mac, password); time.sleep(1) shell(mac, "input keyevent 4"); time.sleep(1) # email field xml = wait_dump(mac) hits = find_in_dump(xml, "email") if not hits: return {"status": "email_field_missing", "log": log} tap(mac, *hits[0][:2]); time.sleep(2); text(mac, email); time.sleep(1) shell(mac, "input keyevent 4"); time.sleep(1) # submit xml = wait_dump(mac) for btn in ("Continue", "Sign up", "Create account"): hits = find_in_dump(xml, btn) if hits: tap(mac, *hits[0][:2]); time.sleep(10) break else: return {"status": "submit_missing", "log": log} # save account row con = sqlite3.connect(DB) con.execute("""CREATE TABLE IF NOT EXISTS reddit_accounts( mac TEXT PRIMARY KEY, username TEXT, password TEXT, email TEXT, status TEXT, created REAL)""") con.execute("INSERT OR REPLACE INTO reddit_accounts(mac, username, password, email, status, created) " "VALUES(?,?,?,?,'created',?)", (mac, username, password, email, time.time())) con.commit() con.close() return {"status": "created", "username": username, "email": email, "log": log} def verify_email(email, timeout=300): """Poll Gmail IMAP for the Reddit verification email, click the link.""" end = time.time() + timeout while time.time() < end: try: M = imaplib.IMAP4_SSL("imap.gmail.com") M.login(GMAIL, APP_PASS) M.select("INBOX") typ, data = M.search(None, f'(FROM "reddit")') for num in data[0].split(): typ, msg = M.fetch(num, "(RFC822)") body = msg[0][1].decode("utf-8", "ignore") m = re.search(r"https://[^\s\"<>]+verif[^\s\"<>]*", body) if m: link = m.group(0).rstrip(".,;") urllib.request.urlopen(link, timeout=30) M.logout() return True, link[:80] M.logout() except Exception: pass time.sleep(15) return False, "" def run(mac): r = create_reddit(mac) if r["status"] == "created": ok, link = verify_email(r["email"]) r["verified"] = ok if ok: r["status"] = "verified" con = sqlite3.connect(DB) con.execute("UPDATE reddit_accounts SET status='verified' WHERE mac=?", (mac,)) con.commit(); con.close() print(json.dumps(r, indent=1)) if __name__ == "__main__": import sys run(sys.argv[1])