import os import re import uuid import random import socket import json import time import string import shutil import logging import base64 import requests import argparse from datetime import datetime, timezone from urllib.parse import urlparse, urlencode from bs4 import BeautifulSoup from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.backends import default_backend # 假设这些是你本地的依赖 import configure from DrissionPage import ChromiumPage, ChromiumOptions from vs_types import RateLimiteddError, BizLogicError from toolkit.mihomo_tunnel import MihomoTunnel from utils.cloudflare_bypass_for_scraping import CloudflareBypasser # ========================================== # 日志与常量配置 # ========================================== logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%H:%M:%S' ) logger = logging.getLogger("VFSRegistrar") VFS_PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY----- MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAuupFgB+lYIOtSxrRoHzc LmCZKJ6+oSbgqgOPzFMM0TasOeLw0NXEn1XfIzXdx75+tegNKwyIZumoh0yhubKs t59GV321kN0iquYRHrdh3ygfDDHlS9rROQeBqRga0ncSADtbLMrBPqXJjPCoV76y t92towriKoH75BhiazY0mghm4LjmAWrV0u/GNpV3tk9bxbtHEXGaFmxCJqjg+7x6 1e5wXLfvpj9w1QsiSWOSJxLOyICz/9ByxXycQQFdNmjnnnwco9Gt/Mi33NYH71j0 5oXIjklFC4lvJqaqSY5lS7Vwb9oCt9zX9J0Yz4z4e/3V+0jgRnWOFGofyks4FKe2 GQIDAQAB -----END PUBLIC KEY-----""" # ========================================== # 辅助工具类 # ========================================== class VFSHelper: @staticmethod def generate_mobile_number(country_code=353, e164_format=False): if country_code == 353: # Ireland prefix = random.choice(['83', '85', '86', '87', '89']) number = f"{prefix}{''.join([str(random.randint(0, 9)) for _ in range(7)])}" return f"+353{number}" if e164_format else number elif country_code == 44: # UK prefix_second = random.choice(['1', '2', '3', '4', '5', '7', '8', '9']) number = f"7{prefix_second}{''.join([str(random.randint(0, 9)) for _ in range(8)])}" return f"+44{number}" if e164_format else number elif country_code == 86: # China prefixes = ["130", "131", "132", "133", "135", "136", "138", "139", "150", "158", "159", "186"] prefix = random.choice(prefixes) number = f"{prefix}{''.join([str(random.randint(0, 9)) for _ in range(8)])}" return f"+86{number}" if e164_format else number return "".join([str(random.randint(0, 9)) for _ in range(10)]) @staticmethod def generate_password(length=12): chars = string.ascii_letters + string.digits + "@#$%" while True: pwd = ''.join(random.choices(chars, k=length)) if (any(c.islower() for c in pwd) and any(c.isupper() for c in pwd) and any(c.isdigit() for c in pwd) and any(c in "@#$%" for c in pwd)): return pwd @staticmethod def encrypt_password(password: str) -> str: public_key = serialization.load_pem_public_key( VFS_PUBLIC_KEY_PEM.encode(), backend=default_backend() ) ciphertext = public_key.encrypt( password.encode(), padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None ) ) return base64.b64encode(ciphertext).decode() @staticmethod def get_client_source() -> str: timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") payload = f"GA;{timestamp}Z" return VFSHelper.encrypt_password(payload) class BrowserResponse: """标准化浏览器响应""" def __init__(self, result_dict): result_dict = result_dict or {} self.status_code = result_dict.get('status', 0) self.text = result_dict.get('body', '') self.headers = result_dict.get('headers', {}) self.url = result_dict.get('url', '') self._json = None def json(self): if self._json is None: try: self._json = json.loads(self.text) if self.text else {} except json.JSONDecodeError: self._json = {} return self._json # ========================================== # API 交互与代理工具 # ========================================== def load_proxies(pool_name): """从 config/proxies.json 读取对应的代理池""" config_path = os.path.join(os.path.dirname(__file__), 'config', 'proxies.json') try: with open(config_path, 'r', encoding='utf-8') as f: data = json.load(f) proxies = data.get(pool_name, []) if not proxies: raise ValueError(f"代理池 '{pool_name}' 为空或不存在！") return proxies except Exception as e: logger.error(f"读取代理配置文件失败: {e}") exit(1) def upload_account_to_server(account, api_url, api_token): """将注册成功的账号上报到中心服务器""" headers = { 'accept': 'application/json', 'Authorization': f'Bearer {api_token}', 'Content-Type': 'application/json' } payload = { "pool_name": account.get("pool_name"), "username": account.get("username"), "password": account.get("password"), "extra_data": { "country_code": account.get("country_code"), "mission_code": account.get("mission_code"), "phone_country_code": account.get("phone_country_code"), "phone_number": account.get("phone_number"), "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } } try: logger.info(f"Uploading account {account['username']} to server...") resp = requests.post(api_url, json=payload, headers=headers, timeout=10) if resp.status_code == 200: logger.info(f"✅ [API Upload Success] Server responded: {resp.text}") return True else: logger.error(f"❌ [API Upload Failed] Status: {resp.status_code}, Body: {resp.text}") return False except Exception as e: logger.error(f"❌ [API Upload Error]: {e}") return False # ========================================== # 核心自动化类 # ========================================== class VFSRegistrationBot: def __init__(self, vfs_url, proxy, email_api_token, master_email): self.vfs_url = vfs_url self.proxy_config = proxy self.email_api_token = email_api_token self.master_email = master_email self.page = None self.tunnel = None self.instance_id = uuid.uuid4().hex[:8] self.workspace = os.path.abspath(os.path.join("data/temp_browser_data", f"reg_session_{self.instance_id}")) self.active_proxy_url = None # 供 request 请求复用 def _log(self, msg): logger.info(f"[TLS-Reg-{self.instance_id}] {msg}") def _init_browser(self): """初始化浏览器配置""" co = ChromiumOptions() # 查找可用端口 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) port = s.getsockname()[1] co.set_local_port(port) chrome_path = configure.CHROME_PATH or os.getenv("CHROME_BIN") if chrome_path and os.path.exists(chrome_path): co.set_paths(browser_path=chrome_path) if self.proxy_config and self.proxy_config.get("ip"): p = self.proxy_config if p.get('username') and p.get('password'): self._log(f"Starting Proxy Tunnel for {p.get('ip')}...") exit_node = { "name": "ExitNode", "type": p.get('proto'), "server": p.get('ip'), "port": p.get('port'), "username": p.get('username'), "password": p.get('password') } relay_node = random.choice(configure.MIHOMO_RELAY_NODES) if configure.MIHOMO_RELAY_NODES else None mihomo_path = configure.MIHOMO_BIN_PATH or os.getenv("MIHOMO_BIN") if not mihomo_path: raise BizLogicError('Mihomo path is null. Set mihomo bin path in configure or os env') self.tunnel = MihomoTunnel(mihomo_path, exit_node=exit_node, relay_node=relay_node) self.active_proxy_url = self.tunnel.start() self._log(f"Tunnel started at {self.active_proxy_url}") co.set_argument(f'--proxy-server={self.active_proxy_url}') else: self.active_proxy_url = f"{p.get('proto')}://{p.get('ip')}:{p.get('port')}" co.set_argument(f'--proxy-server={self.active_proxy_url}') else: self._log("[WARN] No proxy configured!") co.headless(False) co.set_argument('--no-sandbox') co.set_argument('--disable-gpu') co.set_argument('--disable-dev-shm-usage') co.set_argument('--window-size=1920,1080') co.set_argument('--disable-blink-features=AutomationControlled') self.page = ChromiumPage(co) self.page.set.timeouts(15) def _perform_js_fetch(self, method, url, headers=None, data=None, json_data=None, retry_count=0): """注入JS执行Fetch请求，绕过部分指纹检测""" if not self.page: raise BizLogicError("Browser not initialized") if retry_count > 3: raise BizLogicError("Max retries exceeded for request") fetch_options = { "method": method.upper(), "headers": headers or {}, "credentials": "include" } if json_data: fetch_options['body'] = json.dumps(json_data) fetch_options['headers']['Content-Type'] = 'application/json' elif data: fetch_options['body'] = urlencode(data) if isinstance(data, dict) else str(data) fetch_options['headers']['Content-Type'] = 'application/x-www-form-urlencoded' js_script = f""" const url = "{url}"; const options = {json.dumps(fetch_options)}; return fetch(url, options) .then(async response => {{ const text = await response.text(); const headers = {{}}; response.headers.forEach((value, key) => headers[key] = value); return {{ status: response.status, body: text, headers: headers, url: response.url }}; }}) .catch(err => ({{ status: 0, body: err.toString() }})); """ try: res_dict = self.page.run_js(js_script, timeout=60) resp = BrowserResponse(res_dict) if resp.status_code == 200: return resp if resp.status_code == 403 and ("cloudflare" in resp.text.lower() or "Just a moment" in resp.text): logger.warning(f"Cloudflare 403 detected. Retrying ({retry_count+1})...") new_token = self._refresh_turnstile() if new_token and json_data and "captcha_api_key" in json_data: json_data["captcha_api_key"] = new_token return self._perform_js_fetch(method, url, headers, data, json_data, retry_count + 1) if resp.status_code == 429: raise RateLimiteddError(f"Rate Limit: {resp.text[:100]}") return resp except Exception as e: logger.error(f"JS Execution Error: {e}") raise BizLogicError(f"Fetch failed: {e}") def _handle_cookie_banner(self): try: self.page.run_js(""" var btn = document.getElementById('onetrust-accept-btn-handler'); if(btn) { btn.click(); return true; } var banner = document.getElementById('onetrust-banner-sdk'); if(banner) { banner.remove(); return true; } """) except: pass def _refresh_turnstile(self): self._log("Attempting to refresh Turnstile token...") try: self.page.run_js('try{window.turnstile.reset()}catch(e){}') cf_bypasser = CloudflareBypasser(self.page, log=True) for i in range(30): time.sleep(1) try: ele = self.page.ele('@name=cf-turnstile-response') if ele and ele.value: self._log("Turnstile token obtained.") return ele.value except: pass if i > 5: try: cf_bypasser.click_verification_button(is_dfs=False) except: pass except Exception as e: logger.error(f"Turnstile refresh failed: {e}") return None def _wait_for_activation_link(self, username, max_wait_sec=60): self._log(f"Waiting for email to {username}...") start_time = time.time() url = "https://api.text.skin/api/email-authorizations/fetch" headers = { "Authorization": f"Bearer {self.email_api_token}", "Content-Type": "application/json", "Accept": "application/json, text/plain, */*" } while time.time() - start_time < max_wait_sec: try: utc_now_str = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S') params = { "email": self.master_email, "sender": 'donotreply at vfsglobal.com', "recipient": username, "subjectKeywords": 'Welcome', "bodyKeywords": 'ActivateAccount', "sentDate": utc_now_str, "expiry": str(60) } # 已经去掉 proxies=proxies 参数 resp = requests.post(url, headers=headers, params=params, json={}, timeout=60) if resp.status_code != 200: logger.warning(f"Email API returned status {resp.status_code}.") time.sleep(15) continue try: result = resp.json() except json.JSONDecodeError: time.sleep(15) continue if result.get('code') != 0: logger.debug(f"API Message: {result.get('message')}") else: content = result.get('data', {}).get('body', "") if content: soup = BeautifulSoup(content, "html.parser") link = soup.find("a", string="ActivateAccount") if link: return link["href"].replace(" ", "").replace("\n", "").replace("\r", "").strip() except Exception as e: logger.warning(f"Error fetching email: {e}") time.sleep(15) self._log("Checking email again...") return None def register(self, account): try: self._init_browser() self._log(f"Opening {self.vfs_url}") self.page.set.timeouts(page_load=30, script=30) try: self.page.get(self.vfs_url, retry=0, timeout=30) except Exception: logger.warning("Page load timed out. Stopping loading manually.") self.page.stop_loading() cf_token = None cf_bypasser = CloudflareBypasser(self.page, log=True) for _ in range(40): time.sleep(1) if "page-not-found" in self.page.url: logger.error("❌ [BLOCKED] Redirected to 'Page Not Found'. Aborting.") return False if "403" in self.page.title and "Just a moment" not in self.page.title: logger.error("❌ [BLOCKED] 403 Forbidden detected. Aborting.") return False self._handle_cookie_banner() try: ele = self.page.ele('@name=cf-turnstile-response') if ele and ele.value: cf_token = ele.value break except: pass try: cf_bypasser.click_verification_button(is_dfs=False) except: pass if not cf_token: raise BizLogicError("Failed to obtain initial Cloudflare token") post_data = { 'emailid': account['username'], 'password': VFSHelper.encrypt_password(account['password']), 'confirmPassword': VFSHelper.encrypt_password(account['password']), 'processPerDataAgreed': True, 'intTransPerDataAgreed': True, 'termAndConditionAgreed': True, 'missioncode': account['mission_code'], 'countrycode': account['country_code'], 'languageCode': 'en', 'dialcode': str(account['phone_country_code']), 'contact': account['phone_number'], 'captcha_version': 'cloudflare-v1', 'captcha_api_key': cf_token, 'cultureCode': 'en-US', 'IsSpecialUser': False, } headers = { 'content-type': 'application/json;charset=utf-8', 'accept': 'application/json, text/plain, */*', 'route': f"{account['country_code']}/en/{account['mission_code']}", 'clientsource': VFSHelper.get_client_source(), } self._log(f"Submitting registration for {account['username']}") country_code = account.get('country_code', '').lower() if country_code == 'chn': api_domain = 'https://lift-apicn.vfsglobal.com' else: api_domain = 'https://lift-api.vfsglobal.com' register_endpoint = f"{api_domain}/user/registration" self._log(f"Using API Endpoint: {register_endpoint}") resp = self._perform_js_fetch( 'POST', register_endpoint, headers=headers, json_data=post_data ) self._log(f'register resp={resp.text}') resp_data = resp.json() if resp_data.get("code") == "200": self._log("Registration API success. Waiting for email...") activate_link = self._wait_for_activation_link(account['username']) if activate_link: self._log(f"Activating account: {activate_link}") activate_tab = self.page.new_tab(activate_link) try: success_ele = activate_tab.ele('Activation Successful', timeout=30) if success_ele: logger.info(f"✅ Account {account['username']} activated successfully.") return True else: logger.error("Activation verification failed. Success text not found.") return False finally: activate_tab.close() else: logger.error("Timeout waiting for activation email.") else: logger.error(f"Registration failed: {resp_data}") except Exception as e: logger.error(f"Registration process exception: {e}", exc_info=True) return False def cleanup(self): self._log("Cleaning up resources...") if self.page: try: self.page.quit() except: pass if self.tunnel: try: self.tunnel.stop() except: pass if os.path.exists(self.workspace): time.sleep(1) shutil.rmtree(self.workspace, ignore_errors=True) # ========================================== # 主流程 & 命令行参数解析 # ========================================== def parse_arguments(): parser = argparse.ArgumentParser(description="VFS Global 自动注册机器人") # 核心参数 parser.add_argument('--target', type=int, default=20, help="需要注册的目标账号数量") parser.add_argument('--vfs-url', type=str, required=True, help="VFS 注册页面的完整 URL, 如: https://visa.vfsglobal.com/chn/en/aut/register") parser.add_argument('--proxy-pool', type=str, default='proxy-cheap', help="使用的代理池名称 (在 proxies.json 中定义)") # 账号生成相关参数 parser.add_argument('--account-prefix', type=str, default="cn_at", help="生成的邮箱前缀") parser.add_argument('--account-pool', type=str, default="cn.at.sentinel", help="上报到服务端的账号池名称") parser.add_argument('--phone-country', type=int, default=86, help="生成的手机号国家代码") parser.add_argument('--email-domain', type=str, default="text.skin", help="邮箱域名") # API 与 上报配置 parser.add_argument('--upload-url', type=str, default="https://api.text.skin/api/account/add", help="账号上报服务器 URL") parser.add_argument('--upload-token', type=str, default="tok_e946329a60ff45ba807f3f41b0e8b7fc", help="账号上报服务器 API Token") parser.add_argument('--email-token', type=str, default="tok_e946329a60ff45ba807f3f41b0e8b7fc", help="获取邮件的 API Token") parser.add_argument('--master-email', type=str, default="hujiarui8@gmail.com", help="接收验证邮件的主邮箱") return parser.parse_args() def generate_account_details(args): """根据命令行参数和 URL 生成账号数据字典""" # 从 URL 中提取 from_country 和 to_country match = re.search(r'vfsglobal\.com/([^/]+)/[^/]+/([^/]+)/register', args.vfs_url) if not match: raise ValueError(f"无法从 URL 解析国家代码: {args.vfs_url}") from_country = match.group(1) to_country = match.group(2) rand_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6)) username = f"{args.account_prefix}_{rand_suffix}@{args.email_domain}" phone = VFSHelper.generate_mobile_number(args.phone_country) return { 'pool_name': args.account_pool, 'country_code': from_country, 'mission_code': to_country, 'phone_country_code': args.phone_country, 'phone_number': phone, 'username': username, 'password': VFSHelper.generate_password(), } def main(): args = parse_arguments() proxies = load_proxies(args.proxy_pool) logger.info(f"[*] 成功加载代理数量: {len(proxies)} 个") success_accounts = [] logger.info(">>> Starting Registration Bot <<<") while len(success_accounts) < args.target: proxy_config = random.choice(proxies) account_detail = generate_account_details(args) logger.info(f"Processing Account: {account_detail['username']}") bot = None try: bot = VFSRegistrationBot( vfs_url=args.vfs_url, proxy=proxy_config, email_api_token=args.email_token, master_email=args.master_email ) is_success = bot.register(account_detail) if is_success: success_accounts.append(account_detail) logger.info(f"Progress: {len(success_accounts)}/{args.target}") upload_account_to_server(account_detail, args.upload_url, args.upload_token) # 实时保存结果，防止中途退出丢失 with open("registered_accounts.json", "w", encoding='utf-8') as f: json.dump(success_accounts, f, indent=4, ensure_ascii=False) else: logger.warning("Registration failed. Retrying with new account details...") time.sleep(5) # 避免请求过于频繁 except Exception as e: logger.error(f"[ERROR] 注册发生致命错误 | 代理 IP: {proxy_config.get('ip')} | 异常信息: {e}") finally: if bot: bot.cleanup() logger.info(f">>> All tasks completed. Registered {len(success_accounts)} accounts. <<<") if __name__ == "__main__": main()