""" VentraIP DNS Config — Stealth Mode ==================================== Uses playwright-stealth to bypass reCAPTCHA v3 bot detection. Keeps browser session alive, waits for 2FA code via file or --code. Usage: python3 ventraip_stealth.py # Login + wait for code + explore python3 ventraip_stealth.py --code X --configure # Login + code + configure all """ import argparse import json import re import sys import time from pathlib import Path from playwright.sync_api import sync_playwright from playwright_stealth import Stealth VENTRAIP_EMAIL = "kinan@protonmail.com" VENTRAIP_PASSWORD = "9iphQKcIv0#2J0Ce" NETLIFY_A_RECORD = "75.2.60.5" MANIFEST_PATH = "/mnt/e/genesis-system/scripts/gold_pipeline/output/deployment_manifest.json" SCREENSHOT_DIR = Path("/mnt/e/genesis-system/scripts/gold_pipeline/output/ventraip_screenshots") OUTPUT_DIR = Path("/mnt/e/genesis-system/scripts/gold_pipeline/output") CODE_FILE = OUTPUT_DIR / "ventraip_2fa_code.txt" SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True) VENTRAIP_DOMAINS = [ "sydneychatbots.com.au", "brisbanechatbots.com.au", "melbournechatbots.com.au", "adelaidechatbots.com.au", "cairnschatbots.com.au", "sunshinecoastchatbots.com.au", "townsvillechatbots.com.au", "darwinchatbots.com.au", "aichatbotsbrisbane.au", "aichatbotsaustralia.au", "tradiechatbots.com.au", "tradiechatbot.com.au", "tradiesvoice.com.au", "tradieautomation.com.au", "aitradie.au", "sparkybots.com.au", "plumberbots.com.au", "hvacbots.com.au", "electricianbotai.au", "solarbotai.au", "mechanicbot.au", "locksmithbot.au", "rooferbot.au", "receptionistai.au", "beautybot.com.au", "propertybotai.com.au", "propertybotai.au", "dentalbotai.au", "clinicaibot.au", "lawyerbot.au", "legalbotai.au", "rentbot.au", "voiceassistant.com.au", "voiceassistant.au", "aivoicereceptionist.au", "phonebot.au", "talkbot.au", "bookingbot.au", ] def get_domain_map(): with open(MANIFEST_PATH) as f: manifest = json.load(f) result = {} for domain in VENTRAIP_DOMAINS: if domain in manifest: match = re.search(r"https://([^.]+)\.netlify\.app", manifest[domain]) if match: result[domain] = match.group(1) return result def wait_for_code(timeout_secs=300) -> str: """Poll CODE_FILE for the 2FA code.""" if CODE_FILE.exists(): CODE_FILE.unlink() print(f"WAITING_FOR_CODE:{CODE_FILE}") sys.stdout.flush() start = time.time() while time.time() - start < timeout_secs: if CODE_FILE.exists(): code = CODE_FILE.read_text().strip() if code: CODE_FILE.unlink() return code time.sleep(0.5) return "" def configure_domain(page, domain: str, netlify_sub: str) -> dict: netlify_target = f"{netlify_sub}.netlify.app" result = {"domain": domain, "netlify": netlify_target, "status": "unknown", "a_record": False, "cname_record": False} dns_url = f"https://vip.ventraip.com.au/domains/{domain}/dns" page.goto(dns_url, timeout=30000) page.wait_for_timeout(3000) content = page.content() url = page.url if "dns" not in url.lower() and "record" not in content.lower(): for alt in [f"https://vip.ventraip.com.au/domain/{domain}/dns", f"https://vip.ventraip.com.au/domains/{domain}"]: page.goto(alt, timeout=30000) page.wait_for_timeout(2000) if "dns" in page.url.lower() or "record" in page.content().lower(): break page.screenshot(path=str(SCREENSHOT_DIR / f"dns_{domain.replace('.','_')}.png")) add_btn = ( page.query_selector('button:has-text("Add Record")') or page.query_selector('button:has-text("Add DNS")') or page.query_selector('a:has-text("Add Record")') or page.query_selector('button:has-text("Add")') ) if not add_btn: result["status"] = "no_add_button" return result # ADD A RECORD add_btn.click() page.wait_for_timeout(1500) type_sel = page.query_selector('select[name*="type"], select[name*="record"]') if type_sel: type_sel.select_option(label="A") page.wait_for_timeout(500) name_inp = page.query_selector('input[name*="name"], input[name*="host"], input[placeholder*="Name"]') if name_inp: name_inp.fill("@") val_inp = page.query_selector('input[name*="content"], input[name*="value"], input[name*="address"], input[name*="data"], input[placeholder*="Content"], input[placeholder*="Value"]') if val_inp: val_inp.fill(NETLIFY_A_RECORD) save_btn = page.query_selector('button:has-text("Save")') or page.query_selector('button:has-text("Add")') or page.query_selector('button:has-text("Create")') or page.query_selector('button[type="submit"]') if save_btn: save_btn.click() page.wait_for_timeout(2000) result["a_record"] = True # ADD CNAME RECORD page.wait_for_timeout(1000) add_btn2 = page.query_selector('button:has-text("Add Record")') or page.query_selector('a:has-text("Add Record")') or page.query_selector('button:has-text("Add")') if add_btn2: add_btn2.click() page.wait_for_timeout(1500) type_sel2 = page.query_selector('select[name*="type"], select[name*="record"]') if type_sel2: type_sel2.select_option(label="CNAME") page.wait_for_timeout(500) name_inp2 = page.query_selector('input[name*="name"], input[name*="host"], input[placeholder*="Name"]') if name_inp2: name_inp2.fill("www") val_inp2 = page.query_selector('input[name*="content"], input[name*="value"], input[name*="target"], input[name*="data"], input[placeholder*="Content"], input[placeholder*="Value"]') if val_inp2: val_inp2.fill(netlify_target) save_btn2 = page.query_selector('button:has-text("Save")') or page.query_selector('button:has-text("Add")') or page.query_selector('button:has-text("Create")') or page.query_selector('button[type="submit"]') if save_btn2: save_btn2.click() page.wait_for_timeout(2000) result["cname_record"] = True result["status"] = "configured" if (result["a_record"] and result["cname_record"]) else "partial" return result def main(): parser = argparse.ArgumentParser() parser.add_argument("--configure", action="store_true") parser.add_argument("--domain", help="Single domain") parser.add_argument("--code", help="Pre-provide 2FA code") args = parser.parse_args() domain_map = get_domain_map() with sync_playwright() as pw: browser = pw.chromium.launch( headless=True, args=[ "--no-sandbox", "--disable-blink-features=AutomationControlled", "--disable-features=IsolateOrigins,site-per-process", "--disable-site-isolation-trials", "--disable-web-security", ], ) context = browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", viewport={"width": 1920, "height": 1080}, locale="en-AU", timezone_id="Australia/Brisbane", color_scheme="light", java_script_enabled=True, ) page = context.new_page() # Apply stealth patches to avoid bot detection stealth = Stealth() stealth.apply_stealth_sync(page) # Additional anti-detection: override navigator properties page.add_init_script(""" // Override webdriver detection Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); // Chrome runtime window.chrome = { runtime: {}, loadTimes: function(){}, csi: function(){} }; // Permissions const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); // Plugins - make it look like a real browser Object.defineProperty(navigator, 'plugins', { get: () => [ { name: 'Chrome PDF Plugin', filename: 'internal-pdf-viewer' }, { name: 'Chrome PDF Viewer', filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai' }, { name: 'Native Client', filename: 'internal-nacl-plugin' }, ], }); // Languages Object.defineProperty(navigator, 'languages', { get: () => ['en-AU', 'en-US', 'en'] }); // Hardware concurrency Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 }); // Device memory Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 }); // Platform Object.defineProperty(navigator, 'platform', { get: () => 'Win32' }); """) # Intercept and log requests for debugging api_calls = [] def on_request(request): try: url = request.url if "recaptcha" in url.lower() or "2fa" in url.lower() or "verify" in url.lower() or "login" in url.lower(): post_data = None try: post_data = request.post_data except Exception: post_data = "(binary)" api_calls.append({ "url": url, "method": request.method, "post_data": str(post_data)[:500] if post_data else None, }) print(f" [NET] {request.method} {url[:120]}") except Exception: pass page.on("request", on_request) # === LOGIN === print("[LOGIN] Navigating to VIPcontrol (stealth mode)...") page.goto("https://vip.ventraip.com.au/login/", timeout=60000) page.wait_for_timeout(3000) # Simulate human-like behavior: move mouse, wait page.mouse.move(500, 400) page.wait_for_timeout(500) page.mouse.move(800, 300) page.wait_for_timeout(300) # Fill credentials with human-like typing delays email_input = page.query_selector('input[name="email"], input[type="email"]') if email_input: email_input.click() page.wait_for_timeout(200) email_input.type(VENTRAIP_EMAIL, delay=50) page.wait_for_timeout(300) pass_input = page.query_selector('input[name="password"], input[type="password"]') if pass_input: pass_input.click() page.wait_for_timeout(200) pass_input.type(VENTRAIP_PASSWORD, delay=30) page.wait_for_timeout(500) page.screenshot(path=str(SCREENSHOT_DIR / "stealth_login_filled.png")) # Click login with human-like behavior btn = page.query_selector('button:has-text("Login")') or page.query_selector('button[type="submit"]') if btn: btn.hover() page.wait_for_timeout(200) btn.click() print("[LOGIN] Waiting for 2FA page...") page.wait_for_timeout(6000) page.screenshot(path=str(SCREENSHOT_DIR / "stealth_after_login.png")) content = page.content() if "two-factor" in content.lower() or "authentication" in content.lower(): print("[2FA] Code required. Check kinan@protonmail.com") if args.code: code = args.code else: code = wait_for_code(timeout_secs=300) if not code: print("[2FA] No code received within timeout") browser.close() return print(f"[2FA] Entering code: {code}") # Find code input code_input = ( page.query_selector('input[placeholder*="Authentication"]') or page.query_selector('input[placeholder*="Code"]') or page.query_selector('input[name*="code"]') or page.query_selector('input[name*="token"]') ) if not code_input: inputs = page.query_selector_all('input[type="text"], input[type="number"], input:not([type])') for inp in inputs: if inp.is_visible(): code_input = inp break if code_input: # Human-like code entry code_input.click() page.wait_for_timeout(300) code_input.type(code, delay=80) page.wait_for_timeout(500) page.screenshot(path=str(SCREENSHOT_DIR / "stealth_code_entered.png")) # Wait a moment before clicking verify (human behavior) page.wait_for_timeout(800) verify_btn = ( page.query_selector('button:has-text("Verify")') or page.query_selector('button:has-text("Login")') or page.query_selector('button[type="submit"]') ) if verify_btn: verify_btn.hover() page.wait_for_timeout(300) verify_btn.click() page.wait_for_timeout(10000) page.screenshot(path=str(SCREENSHOT_DIR / "stealth_after_2fa.png")) url = page.url print(f"[2FA] URL after verify: {url}") if "incorrect" in page.content().lower(): print("[2FA] FAILED: Incorrect code") # Dump API calls for debugging if api_calls: print(f"\n[DEBUG] Captured {len(api_calls)} relevant requests:") for call in api_calls: print(f" {call['method']} {call['url'][:120]}") if call['post_data']: print(f" POST: {call['post_data'][:300]}") api_path = OUTPUT_DIR / "stealth_api_calls.json" with open(api_path, "w") as f: json.dump(api_calls, f, indent=2, default=str) browser.close() return if "login" in url.lower() and "dashboard" not in url.lower(): print("[2FA] May have failed. Check stealth_after_2fa.png") browser.close() return print("[2FA] LOGIN SUCCESSFUL!") else: print("[2FA] Could not find code input") browser.close() return elif "login" not in page.url.lower(): print("[LOGIN] Already authenticated!") else: print("[LOGIN] Unexpected state") page.screenshot(path=str(SCREENSHOT_DIR / "stealth_unexpected.png")) browser.close() return # === AUTHENTICATED — NOW WORK === if args.configure or args.domain: if args.domain: if args.domain not in domain_map: print(f"Domain {args.domain} not in manifest") browser.close() return domains = {args.domain: domain_map[args.domain]} else: domains = domain_map print(f"\n{'='*60}") print(f"CONFIGURING DNS — {len(domains)} domains") print(f"{'='*60}\n") results = [] for i, (domain, netlify_sub) in enumerate(domains.items(), 1): print(f"[{i}/{len(domains)}] {domain}") result = configure_domain(page, domain, netlify_sub) results.append(result) print(f" → {result['status']} | A={result['a_record']} CNAME={result['cname_record']}") if i < len(domains): time.sleep(1.5) results_path = OUTPUT_DIR / "ventraip_dns_results.json" with open(results_path, "w") as f: json.dump(results, f, indent=2) configured = sum(1 for r in results if r["status"] == "configured") partial = sum(1 for r in results if r["status"] == "partial") failed = len(results) - configured - partial print(f"\n{'='*60}") print(f"DONE: {configured} configured, {partial} partial, {failed} failed") print(f"{'='*60}") else: first = list(domain_map.keys())[0] print(f"\n[EXPLORE] Navigating to DNS for {first}...") for url_pattern in [ f"https://vip.ventraip.com.au/domains/{first}/dns", f"https://vip.ventraip.com.au/domain/{first}/dns", f"https://vip.ventraip.com.au/domains/{first}", "https://vip.ventraip.com.au/domains", ]: page.goto(url_pattern, timeout=30000) page.wait_for_timeout(3000) page.screenshot(path=str(SCREENSHOT_DIR / f"explore_{url_pattern.split('/')[-1].replace('.','_')}.png")) print(f" {url_pattern} → {page.url}") content = page.content() if "dns" in page.url.lower() or "record" in content.lower(): print(" DNS page found!") html_path = SCREENSHOT_DIR / f"explore_{first.replace('.','_')}.html" html_path.write_text(page.content()[:100000], encoding="utf-8") print(f" HTML: {html_path}") break # Save API calls if api_calls: api_path = OUTPUT_DIR / "stealth_api_calls.json" with open(api_path, "w") as f: json.dump(api_calls, f, indent=2, default=str) browser.close() if __name__ == "__main__": main()