From 478efecc32e46e4e22a388f5acb68f6cf8f041e9 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:13:09 +0000 Subject: [PATCH] fix: complete remaining registration hardening --- .../workflows/temp-remaining-hardening.yml | 30 - cpa_export.py | 8 +- cpa_xai/browser_confirm.py | 35 +- cpa_xai/mint.py | 4 + cpa_xai/oauth_device.py | 94 +- grok_register_ttk.py | 670 ++++++------- registration_flow.py | 323 +++---- tools/apply_remaining_hardening.py | 890 ------------------ tools/fix_remaining_patch_oidc_scope.py | 23 - 9 files changed, 501 insertions(+), 1576 deletions(-) delete mode 100644 .github/workflows/temp-remaining-hardening.yml delete mode 100644 tools/apply_remaining_hardening.py delete mode 100644 tools/fix_remaining_patch_oidc_scope.py diff --git a/.github/workflows/temp-remaining-hardening.yml b/.github/workflows/temp-remaining-hardening.yml deleted file mode 100644 index f9500d2..0000000 --- a/.github/workflows/temp-remaining-hardening.yml +++ /dev/null @@ -1,30 +0,0 @@ -name: Temporary remaining hardening - -on: - push: - paths: - - tools/apply_remaining_hardening.py - - tools/fix_remaining_patch_oidc_scope.py - - .github/workflows/temp-remaining-hardening.yml - workflow_dispatch: - -permissions: - contents: write - -jobs: - migrate: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - name: Repair OIDC migration scope - run: python tools/fix_remaining_patch_oidc_scope.py - - name: Apply source migration - run: python tools/apply_remaining_hardening.py - - name: Commit source changes and remove temporary files - run: | - git config user.name "github-actions[bot]" - git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - git rm -f tools/apply_remaining_hardening.py tools/fix_remaining_patch_oidc_scope.py .github/workflows/temp-remaining-hardening.yml - git add -A - git commit -m "fix: complete remaining registration hardening" - git push diff --git a/cpa_export.py b/cpa_export.py index 6d4a479..b44b5d2 100644 --- a/cpa_export.py +++ b/cpa_export.py @@ -86,6 +86,8 @@ def export_cpa_xai_for_account( proxy = str(cfg.get("cpa_proxy") or cfg.get("proxy") or "").strip() headless = bool(cfg.get("cpa_headless", False)) timeout = float(cfg.get("cpa_mint_timeout_sec") or 300) + request_timeout = float(cfg.get("cpa_oidc_request_timeout_sec") or 15) + poll_timeout = float(cfg.get("cpa_oidc_poll_timeout_sec") or 15) base_url = str(cfg.get("cpa_base_url") or "https://cli-chat-proxy.grok.com/v1").strip() force_standalone = bool(cfg.get("cpa_force_standalone", True)) cookie_inject = bool(cfg.get("cpa_mint_cookie_inject", True)) @@ -133,6 +135,8 @@ def export_cpa_xai_for_account( recycle_every=15, log=_log, cancel=cancel_callback, + request_timeout_sec=request_timeout, + poll_timeout_sec=poll_timeout, ) if result.get("ok") and result.get("path") and bool(cfg.get("cpa_copy_to_hotload", False)) and hotload_dir: try: @@ -154,6 +158,6 @@ def export_cpa_xai_for_account( try: with open(str(fail_path), "a", encoding="utf-8") as handle: handle.write("%s----%s----%s\n" % (email, result.get("error") or "unknown", int(time.time()))) - except Exception: - pass + except Exception as exc: + log("[cpa] failed to persist failure record: %s" % exc) return result diff --git a/cpa_xai/browser_confirm.py b/cpa_xai/browser_confirm.py index 7a69890..7322568 100644 --- a/cpa_xai/browser_confirm.py +++ b/cpa_xai/browser_confirm.py @@ -193,6 +193,7 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False, resolved = resolve_proxy(proxy) proxy_bridge = None + browser = None chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger) try: if chrome_proxy: @@ -200,19 +201,17 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False, logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy)) else: logger("browser proxy=(none)") - browser = Chromium(options) if proxy_bridge is not None: - try: - setattr(browser, "_cpa_proxy_bridge", proxy_bridge) - except Exception: - pass - _register_mint_browser(browser) + setattr(browser, "_cpa_proxy_bridge", proxy_bridge) page = browser.latest_tab + _register_mint_browser(browser) logger("standalone chromium started") return browser, page except Exception: - if proxy_bridge is not None: + if browser is not None: + close_standalone(browser) + elif proxy_bridge is not None: try: proxy_bridge.stop() except Exception: @@ -846,6 +845,8 @@ def mint_with_browser( cookies: Any = None, reuse_browser: bool = True, recycle_every: int = 15, + request_timeout_sec: float = 15.0, + poll_timeout_sec: float = 15.0, ): from .oauth_device import OAuthDeviceError, poll_device_token, request_device_code from .proxyutil import proxy_log_label, resolve_proxy, set_runtime_proxy @@ -858,19 +859,12 @@ def mint_with_browser( set_runtime_proxy(resolved or None) success = False try: - last_error = None - session = None - for attempt in range(1, 4): - try: - session = request_device_code(proxy=resolved or None) - last_error = None - break - except Exception as exc: - last_error = exc - logger("request_device_code attempt %s/3 failed: %s" % (attempt, exc)) - _sleep(1.5 * attempt) - if session is None: - raise last_error or RuntimeError("request_device_code failed") + session = request_device_code( + proxy=resolved or None, + timeout=float(request_timeout_sec), + cancel=cancel, + retries=2, + ) logger("device user_code=%s expires_in=%s proxy=%s" % (session.user_code, session.expires_in, proxy_log_label(resolved) or "(none)")) if work_page is None: own_browser, work_page, owned = acquire_mint_browser( @@ -910,6 +904,7 @@ def mint_with_browser( log=logger, cancel=combined_cancel, proxy=resolved or None, + timeout=float(poll_timeout_sec), ) token_box["token"] = result stop_event.set() diff --git a/cpa_xai/mint.py b/cpa_xai/mint.py index 719f5ec..c434f07 100644 --- a/cpa_xai/mint.py +++ b/cpa_xai/mint.py @@ -20,6 +20,8 @@ def mint_and_export( recycle_every=15, log=None, cancel=None, + request_timeout_sec=15.0, + poll_timeout_sec=15.0, ): logger = log or (lambda message: None) email = str(email or "").strip() @@ -40,6 +42,8 @@ def mint_and_export( cookies=cookies, reuse_browser=bool(reuse_browser), recycle_every=int(recycle_every or 0), + request_timeout_sec=float(request_timeout_sec), + poll_timeout_sec=float(poll_timeout_sec), ) except Exception as exc: logger("mint failed: %s" % exc) diff --git a/cpa_xai/oauth_device.py b/cpa_xai/oauth_device.py index f1a1461..7783786 100644 --- a/cpa_xai/oauth_device.py +++ b/cpa_xai/oauth_device.py @@ -72,34 +72,56 @@ def _validate_endpoint(raw_url, field_name): return value -def discover(proxy=None, timeout=30.0): - opener = _build_opener(proxy) +def discover(proxy=None, timeout=30.0, cancel=None, retries=2): request = urllib.request.Request( DISCOVERY_URL, method="GET", headers={"Accept": "application/json", "User-Agent": "grok-register-cpa/1.0"}, ) - try: - with opener.open(request, timeout=timeout) as response: - body = response.read().decode("utf-8", errors="replace") - status = int(getattr(response, "status", 200) or 200) - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8", errors="replace") - raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (exc.code, body)) - except Exception as exc: - raise OAuthDeviceError("xAI discovery request failed: %s" % exc) - if status != 200: - raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (status, body)) - try: - payload = json.loads(body) - except Exception as exc: - raise OAuthDeviceError("xAI discovery parse failed: %s" % exc) - return { - "device_authorization_endpoint": _validate_endpoint( - payload.get("device_authorization_endpoint"), "device_authorization_endpoint" - ), - "token_endpoint": _validate_endpoint(payload.get("token_endpoint"), "token_endpoint"), - } + last_error = None + for attempt in range(max(int(retries), 0) + 1): + _check_cancel(cancel) + opener = _build_opener(proxy) + try: + with opener.open(request, timeout=float(timeout)) as response: + body = response.read().decode("utf-8", errors="replace") + status = int(getattr(response, "status", 200) or 200) + _check_cancel(cancel) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (exc.code, body)) + except Exception as exc: + last_error = exc + if not _is_transient_net_error(exc) or attempt >= int(retries): + raise OAuthDeviceError("xAI discovery request failed: %s" % exc) + _sleep_with_cancel(1.0 * (attempt + 1), cancel) + continue + if status != 200: + raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (status, body)) + try: + payload = json.loads(body) + except Exception as exc: + raise OAuthDeviceError("xAI discovery parse failed: %s" % exc) + return { + "device_authorization_endpoint": _validate_endpoint( + payload.get("device_authorization_endpoint"), "device_authorization_endpoint" + ), + "token_endpoint": _validate_endpoint(payload.get("token_endpoint"), "token_endpoint"), + } + raise OAuthDeviceError("xAI discovery failed: %s" % last_error) + + +def _check_cancel(cancel): + if cancel and cancel(): + raise OAuthDeviceError("cancelled") + + +def _sleep_with_cancel(seconds, cancel=None): + deadline = time.time() + max(float(seconds), 0.0) + while time.time() < deadline: + _check_cancel(cancel) + time.sleep(min(0.2, max(deadline - time.time(), 0.0))) + _check_cancel(cancel) def _is_transient_net_error(exc): @@ -145,7 +167,7 @@ def _is_transient_net_error(exc): return False -def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5): +def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5, cancel=None): data = urllib.parse.urlencode(form).encode("utf-8") request = urllib.request.Request( url, @@ -159,11 +181,13 @@ def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5): ) last_error = None for attempt in range(max(int(retries), 0) + 1): + _check_cancel(cancel) opener = _build_opener(proxy) try: with opener.open(request, timeout=timeout) as response: body = response.read().decode("utf-8", errors="replace") status = int(getattr(response, "status", 200) or 200) + _check_cancel(cancel) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") status = int(exc.code) @@ -171,7 +195,7 @@ def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5): last_error = exc if not _is_transient_net_error(exc) or attempt >= int(retries): raise - time.sleep(float(retry_sleep) * (attempt + 1)) + _sleep_with_cancel(float(retry_sleep) * (attempt + 1), cancel) continue try: return status, json.loads(body) @@ -182,8 +206,9 @@ def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5): raise OAuthDeviceError("form request failed without response") -def request_device_code(client_id=CLIENT_ID, scope=SCOPE, timeout=30.0, proxy=None): - discovery = discover(proxy=proxy, timeout=timeout) +def request_device_code(client_id=CLIENT_ID, scope=SCOPE, timeout=15.0, proxy=None, cancel=None, retries=2): + discovery = discover(proxy=proxy, timeout=timeout, cancel=cancel, retries=retries) + _check_cancel(cancel) device_endpoint = discovery["device_authorization_endpoint"] token_endpoint = discovery["token_endpoint"] status, payload = _post_form( @@ -191,9 +216,11 @@ def request_device_code(client_id=CLIENT_ID, scope=SCOPE, timeout=30.0, proxy=No {"client_id": client_id, "scope": scope}, timeout=timeout, proxy=proxy, - retries=2, + retries=retries, retry_sleep=1.0, + cancel=cancel, ) + _check_cancel(cancel) if status != 200 or not isinstance(payload, dict): raise OAuthDeviceError("device code request failed HTTP %s: %r" % (status, payload)) device_code = str(payload.get("device_code") or "").strip() @@ -216,14 +243,6 @@ def request_device_code(client_id=CLIENT_ID, scope=SCOPE, timeout=30.0, proxy=No ) -def _sleep_with_cancel(seconds, cancel=None): - deadline = time.time() + max(float(seconds), 0.0) - while time.time() < deadline: - if cancel and cancel(): - raise OAuthDeviceError("cancelled") - time.sleep(min(0.2, max(deadline - time.time(), 0.0))) - - def poll_device_token( device_code, token_endpoint, @@ -251,10 +270,11 @@ def poll_device_token( "device_code": str(device_code).strip(), "client_id": client_id, }, - timeout=min(float(timeout), 5.0), + timeout=float(timeout), proxy=proxy, retries=0, retry_sleep=1.0, + cancel=cancel, ) net_streak = 0 except Exception as exc: diff --git a/grok_register_ttk.py b/grok_register_ttk.py index ba816a9..48ba4e7 100755 --- a/grok_register_ttk.py +++ b/grok_register_ttk.py @@ -37,6 +37,7 @@ import socketserver import ssl import urllib.parse import tempfile +import traceback os.environ.setdefault("TK_SILENCE_DEPRECATION", "1") @@ -90,6 +91,9 @@ DEFAULT_CONFIG = { "cpa_force_standalone": True, "cpa_mint_timeout_sec": 300, "cpa_mint_cookie_inject": True, + "cpa_oidc_request_timeout_sec": 15, + "cpa_oidc_poll_timeout_sec": 15, + "grok2api_allow_legacy_full_save": False, } config = DEFAULT_CONFIG.copy() @@ -109,23 +113,103 @@ class ConfigError(RuntimeError): pass +class RemoteTokenCompatibilityError(RuntimeError): + pass + + +class RemoteTokenRequestError(RuntimeError): + pass + + +def log_exception(context, exc, log_callback=None): + message = f"{context}: {exc.__class__.__name__}: {exc}" + if log_callback: + log_callback(f"[!] {message}") + else: + print(f"[!] {message}", file=sys.stderr) + return message + + +def _require_bool(cfg, key): + value = cfg.get(key) + if type(value) is not bool: + raise ConfigError(f"配置项 {key} 必须是布尔值 true/false") + return value + + +def _require_int(cfg, key, minimum, maximum): + value = cfg.get(key) + if type(value) is not int: + raise ConfigError(f"配置项 {key} 必须是整数") + if not minimum <= value <= maximum: + raise ConfigError(f"配置项 {key} 必须在 {minimum} 到 {maximum} 之间") + return value + + +def _require_string(cfg, key, path=False): + value = cfg.get(key) + if not isinstance(value, str): + raise ConfigError(f"配置项 {key} 必须是字符串") + value = value.strip() if key not in ("user_agent",) else value + if "\x00" in value: + raise ConfigError(f"配置项 {key} 包含非法空字符") + if path and value: + os.path.expanduser(value) + return value + + +def validate_config(raw): + if not isinstance(raw, dict): + raise ConfigError("config root must be a JSON object") + cfg = {**DEFAULT_CONFIG, **raw} + bool_keys = ( + "enable_nsfw", "grok2api_auto_add_local", "grok2api_auto_add_remote", + "grok2api_allow_legacy_full_save", "cpa_export_enabled", + "cpa_copy_to_hotload", "cpa_headless", "cpa_force_standalone", + "cpa_mint_cookie_inject", + ) + for key in bool_keys: + cfg[key] = _require_bool(cfg, key) + cfg["register_count"] = _require_int(cfg, "register_count", 1, 2500) + cfg["cpa_mint_timeout_sec"] = _require_int(cfg, "cpa_mint_timeout_sec", 30, 1800) + cfg["cpa_oidc_request_timeout_sec"] = _require_int(cfg, "cpa_oidc_request_timeout_sec", 3, 120) + cfg["cpa_oidc_poll_timeout_sec"] = _require_int(cfg, "cpa_oidc_poll_timeout_sec", 3, 120) + string_keys = tuple(key for key, value in DEFAULT_CONFIG.items() if isinstance(value, str)) + path_keys = {"grok2api_local_token_file", "api_reverse_tools", "cpa_auth_dir", "cpa_hotload_dir"} + for key in string_keys: + cfg[key] = _require_string(cfg, key, path=key in path_keys) + enums = { + "email_provider": {"duckmail", "yyds", "cloudflare", "cloudmail"}, + "cloudflare_auth_mode": {"query-key", "bearer", "x-api-key", "x-admin-auth", "none"}, + "grok2api_pool_name": {"ssoBasic", "ssoSuper"}, + } + for key, allowed in enums.items(): + value = cfg.get(key, DEFAULT_CONFIG.get(key, "")) + if value not in allowed: + raise ConfigError(f"配置项 {key} 的值无效: {value!r}; 允许值: {sorted(allowed)}") + cfg[key] = value + return cfg + + def load_config(): global config if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: loaded = json.load(f) - if not isinstance(loaded, dict): - raise ValueError("config root must be a JSON object") - config = {**DEFAULT_CONFIG, **loaded} + config = validate_config(loaded) + except ConfigError: + raise except Exception as exc: raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc else: - config = DEFAULT_CONFIG.copy() + config = validate_config(DEFAULT_CONFIG.copy()) return config def save_config(): + global config + config = validate_config(config) config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE)) os.makedirs(config_dir, exist_ok=True) fd = None @@ -198,8 +282,6 @@ def warn_runtime_compatibility(): ensure_stable_python_runtime() warn_runtime_compatibility() -load_config() - EXTENSION_PATH = os.path.abspath( os.path.join(os.path.dirname(__file__), "turnstilePatch") ) @@ -819,106 +901,109 @@ def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None): return False base = str(config.get("grok2api_remote_base", "") or "").strip().rstrip("/") app_key = str(config.get("grok2api_remote_app_key", "") or "").strip() - pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() or "ssoBasic" + pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() if not base or not app_key: - if log_callback: - log_callback("[Debug] grok2api 远端未配置 base/app_key，跳过") - return False + raise RemoteTokenRequestError("grok2api 远端未配置 base/app_key") headers = {"Content-Type": "application/json"} query = {"app_key": app_key} - pool_map = {"ssoBasic": "basic", "ssoSuper": "super"} - remote_pool = pool_map.get(pool_name, "basic") + remote_pool = {"ssoBasic": "basic", "ssoSuper": "super"}[pool_name] api_bases = get_grok2api_remote_api_bases(base) - add_errors = [] - # 优先使用 add 接口，避免全量覆盖远端池 + incompatible = [] add_payload = {"tokens": [token], "pool": remote_pool, "tags": ["auto-register"]} for api_base in api_bases: endpoint = f"{api_base}/tokens/add" try: - resp_add = http_post( - endpoint, - headers=headers, - params=query, - json=add_payload, - timeout=30, - ) - resp_add.raise_for_status() + response = http_post(endpoint, headers=headers, params=query, json=add_payload, timeout=30) + except Exception as exc: + raise RemoteTokenRequestError(f"远端 /tokens/add 网络请求失败: {endpoint}: {exc}") from exc + status = int(getattr(response, "status_code", 0) or 0) + if 200 <= status < 300: if log_callback: log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({endpoint})") return True - except Exception as add_exc: - add_errors.append(f"{endpoint}: {add_exc}") - if log_callback: - log_callback(f"[Debug] /tokens/add 写入失败，尝试 /tokens 全量模式: {'; '.join(add_errors)}") - - # 兜底：旧版全量保存接口。必须先成功读取远端旧状态，避免空池覆盖。 - current = {} - fallback_base = api_bases[0] if api_bases else base - loaded_remote_state = False + if status in (404, 405): + incompatible.append(f"{endpoint}: HTTP {status}") + continue + body = str(getattr(response, "text", "") or "")[:300] + raise RemoteTokenRequestError(f"远端 /tokens/add 请求失败，不允许全量回退: {endpoint}: HTTP {status}: {body}") + if not bool(config.get("grok2api_allow_legacy_full_save", False)): + raise RemoteTokenCompatibilityError( + "/tokens/add 不受支持，旧版全量保存默认禁用以避免并发覆盖: " + "; ".join(incompatible) + ) + current = None + fallback_base = None + etag = None load_errors = [] for api_base in api_bases or [base]: + endpoint = f"{api_base}/tokens" try: - resp = http_get(f"{api_base}/tokens", headers=headers, params=query, timeout=20) - if resp.status_code == 200: - payload = resp.json() - if isinstance(payload, dict): - candidate = payload.get("tokens") if "tokens" in payload else payload - if isinstance(candidate, dict): - current = candidate - fallback_base = api_base - loaded_remote_state = True - break - load_errors.append(f"{api_base}/tokens: unexpected payload") - else: - load_errors.append(f"{api_base}/tokens: HTTP {resp.status_code}") + response = http_get(endpoint, headers=headers, params=query, timeout=20) except Exception as exc: - load_errors.append(f"{api_base}/tokens: {exc}") - if not loaded_remote_state: - raise RuntimeError("无法安全读取远端 token 池，拒绝执行全量覆盖: " + "; ".join(load_errors)) + raise RemoteTokenRequestError(f"旧版远端池读取网络失败: {endpoint}: {exc}") from exc + status = int(getattr(response, "status_code", 0) or 0) + if status != 200: + load_errors.append(f"{endpoint}: HTTP {status}") + continue + payload = response.json() + candidate = payload.get("tokens") if isinstance(payload, dict) and "tokens" in payload else payload + if not isinstance(candidate, dict): + load_errors.append(f"{endpoint}: unexpected payload") + continue + current = candidate + fallback_base = api_base + response_headers = getattr(response, "headers", {}) or {} + etag = response_headers.get("ETag") or response_headers.get("etag") + break + if current is None or fallback_base is None: + raise RemoteTokenRequestError("无法安全读取旧版远端 token 池: " + "; ".join(load_errors)) pool = current.get(pool_name) if pool is None: pool = [] elif not isinstance(pool, list): - raise RuntimeError(f"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖") - existing = set() - for item in pool: - if isinstance(item, str): - existing.add(_normalize_sso_token(item)) - elif isinstance(item, dict): - existing.add(_normalize_sso_token(item.get("token", ""))) + raise RemoteTokenRequestError(f"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖") + existing = { + _normalize_sso_token(item if isinstance(item, str) else item.get("token", "")) + for item in pool if isinstance(item, (str, dict)) + } if token not in existing: pool.append({"token": token, "tags": ["auto-register"], "note": email}) current[pool_name] = pool - save_errors = [] - save_bases = [] - for item in [fallback_base, *(api_bases or [base])]: - if item and item not in save_bases: - save_bases.append(item) - for api_base in save_bases: - try: - resp2 = http_post(f"{api_base}/tokens", headers=headers, params=query, json=current, timeout=30) - resp2.raise_for_status() - if log_callback: - log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({api_base}/tokens)") - return True - except Exception as save_exc: - save_errors.append(f"{api_base}/tokens: {save_exc}") - raise RuntimeError(f"grok2api 远端 /tokens 全量模式写入失败: {'; '.join(save_errors)}") + save_headers = dict(headers) + if etag: + save_headers["If-Match"] = etag + elif log_callback: + log_callback("[!] 旧版远端接口未提供 ETag；已由显式配置允许，但仍不建议多实例并发") + endpoint = f"{fallback_base}/tokens" + try: + response = http_post(endpoint, headers=save_headers, params=query, json=current, timeout=30) + except Exception as exc: + raise RemoteTokenRequestError(f"旧版远端池保存网络失败: {endpoint}: {exc}") from exc + status = int(getattr(response, "status_code", 0) or 0) + if not 200 <= status < 300: + raise RemoteTokenRequestError(f"旧版远端池保存失败: {endpoint}: HTTP {status}") + if log_callback: + log_callback(f"[+] 已写入 grok2api 远端池（旧版兼容）: {pool_name} ({endpoint})") + return True def add_token_to_grok2api_pools(raw_token, email="", log_callback=None): - if config.get("grok2api_auto_add_local", True): + result = { + "local": {"enabled": bool(config.get("grok2api_auto_add_local", False)), "ok": None, "error": None}, + "remote": {"enabled": bool(config.get("grok2api_auto_add_remote", False)), "ok": None, "error": None}, + } + if result["local"]["enabled"]: try: - add_token_to_grok2api_local_pool(raw_token, email=email, log_callback=log_callback) + result["local"]["ok"] = bool(add_token_to_grok2api_local_pool(raw_token, email=email, log_callback=log_callback)) except Exception as exc: - if log_callback: - log_callback(f"[Debug] 写入 grok2api 本地池失败: {exc}") - if config.get("grok2api_auto_add_remote", False): + result["local"]["ok"] = False + result["local"]["error"] = log_exception("写入 grok2api 本地池失败", exc, log_callback) + if result["remote"]["enabled"]: try: - add_token_to_grok2api_remote_pool(raw_token, email=email, log_callback=log_callback) + result["remote"]["ok"] = bool(add_token_to_grok2api_remote_pool(raw_token, email=email, log_callback=log_callback)) except Exception as exc: - if log_callback: - log_callback(f"[Debug] 写入 grok2api 远端池失败: {exc}") + result["remote"]["ok"] = False + result["remote"]["error"] = log_exception("写入 grok2api 远端池失败", exc, log_callback) + return result def apply_browser_proxy_option(options, proxy): @@ -3120,6 +3205,84 @@ def maybe_export_cpa_xai_after_success(email, password, sso="", log_callback=Non return result + +def _save_mail_credential(email, credential, log_callback=None): + path = os.path.join(os.path.dirname(__file__), "mail_credentials.txt") + try: + with open(path, "a", encoding="utf-8") as handle: + handle.write(f"{email}\t{credential}\n") + handle.flush() + return True + except Exception as exc: + log_exception("保存邮箱凭据失败", exc, log_callback) + return False + + +def _append_account_line(path, email, password, sso): + with open(path, "a", encoding="utf-8") as handle: + handle.write(f"{email}----{password}----{sso}\n") + handle.flush() + os.fsync(handle.fileno()) + + +def _queue_unsaved_account(path, payload, error, log_callback=None): + pending_path = path + ".pending.jsonl" + record = dict(payload) + record["save_error"] = str(error) + record["queued_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() + try: + with open(pending_path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, ensure_ascii=False) + "\n") + handle.flush() + os.fsync(handle.fileno()) + try: + os.chmod(pending_path, 0o600) + except Exception: + pass + return True + except Exception as exc: + log_exception("写入账号 pending 队列失败", exc, log_callback) + return False + + +def run_registration_common(count, log_callback, cancel_callback, accounts_output_file, observer): + from registration_flow import RegistrationCallbacks, RegistrationOperations, run_batch + callbacks = RegistrationCallbacks(log=log_callback, cancelled=cancel_callback) + operations = RegistrationOperations( + start_browser=lambda: start_browser(log_callback=log_callback), + restart_browser=lambda: restart_browser(log_callback=log_callback), + browser_missing=lambda: browser is None, + open_signup_page=lambda: open_signup_page(log_callback=log_callback, cancel_callback=cancel_callback), + fill_email_and_submit=lambda: fill_email_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), + save_mail_credential=lambda email, token: _save_mail_credential(email, token, log_callback), + fill_code_and_submit=lambda email, token: fill_code_and_submit(email, token, log_callback=log_callback, cancel_callback=cancel_callback), + fill_profile_and_submit=lambda: fill_profile_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), + wait_for_sso_cookie=lambda: wait_for_sso_cookie(log_callback=log_callback, cancel_callback=cancel_callback), + enable_nsfw=lambda sso: enable_nsfw_for_token(sso, log_callback=log_callback), + persist_account_line=lambda email, password, sso: _append_account_line(accounts_output_file, email, password, sso), + queue_unsaved_result=lambda payload, error: _queue_unsaved_account(accounts_output_file, payload, error, log_callback), + add_tokens=lambda sso, email: add_token_to_grok2api_pools(sso, email=email, log_callback=log_callback), + export_cpa=lambda email, password, sso: maybe_export_cpa_xai_after_success( + email=email, password=password, sso=sso, + log_callback=log_callback, cancel_callback=cancel_callback, + ), + cleanup=lambda reason: cleanup_runtime_memory(log_callback=log_callback, reason=reason), + sleep=lambda seconds: sleep_with_cancel(seconds, cancel_callback), + cancelled_exception=RegistrationCancelled, + retry_exception=AccountRetryNeeded, + ) + return run_batch( + count=count, + callbacks=callbacks, + observer=observer, + ops=operations, + enable_nsfw=bool(config.get("enable_nsfw", True)), + cleanup_interval=MEMORY_CLEANUP_INTERVAL, + max_slot_retry=3, + max_mail_retry=3, + ) + + class GrokRegisterGUI: def __init__(self, root): self.root = root @@ -3133,9 +3296,9 @@ class GrokRegisterGUI: self.results = [] self.stop_requested = False self.ui_queue = queue.Queue() - self._ui_thread_id = threading.get_ident() self.accounts_output_file = "" self.setup_ui() + self.root.after(50, self.process_ui_queue) def setup_ui(self): load_config() @@ -3351,39 +3514,50 @@ class GrokRegisterGUI: self.log("[*] GUI 已就绪，配置已加载") self.log(f"[*] 当前邮箱服务商: {self.email_provider_var.get()} | 注册数量: {self.count_var.get()}") - def _call_ui(self, func, *args): - if threading.get_ident() == self._ui_thread_id: - func(*args) - return + def process_ui_queue(self): try: - self.root.after(0, lambda: func(*args)) - except Exception: + while True: + event = self.ui_queue.get_nowait() + kind = event[0] + if kind == "log": + line = event[1] + self.log_text.insert(tk.END, f"{line}\n") + self.log_text.see(tk.END) + elif kind == "stats": + self.stats_var.set(f"成功: {event[1]} | 失败: {event[2]}") + elif kind == "running": + running = bool(event[1]) + self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL) + self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED) + self.status_var.set("运行中..." if running else "就绪") + self.status_label.config(foreground="blue" if running else "green") + elif kind == "error": + messagebox.showerror(event[1], event[2]) + except queue.Empty: pass - - def _append_log_line(self, line): - self.log_text.insert(tk.END, f"{line}\n") - self.log_text.see(tk.END) + except Exception as exc: + print(f"[!] UI 队列处理失败: {exc}", file=sys.stderr) + finally: + try: + self.root.after(50, self.process_ui_queue) + except Exception: + pass def log(self, message): timestamp = datetime.datetime.now().strftime("%H:%M:%S") line = f"[{timestamp}] {message}" print(line, flush=True) - self._call_ui(self._append_log_line, line) + self.ui_queue.put(("log", line)) def clear_log(self): - self._call_ui(lambda: self.log_text.delete(1.0, tk.END)) + self.log_text.delete(1.0, tk.END) def update_stats(self): - self._call_ui(lambda: self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}")) + self.ui_queue.put(("stats", self.success_count, self.fail_count)) def _set_running_ui(self, running): - self.is_running = running - def apply(): - self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL) - self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED) - self.status_var.set("运行中..." if running else "就绪") - self.status_label.config(foreground="blue" if running else "green") - self._call_ui(apply) + self.is_running = bool(running) + self.ui_queue.put(("running", self.is_running)) def should_stop(self): @@ -3471,148 +3645,31 @@ class GrokRegisterGUI: self.log("[!] 用户停止注册") def run_registration(self, count): + def observer(batch, account, output): + self.success_count = batch.success_count + self.fail_count = batch.fail_count + if account is not None: + self.results.append({"email": account.email, "sso": account.sso, "profile": account.profile, "output": output}) + self.update_stats() try: - start_browser(log_callback=self.log) - self.log("[*] 浏览器已启动") - i = 0 - retry_count_for_slot = 0 - max_slot_retry = 3 - while i < count: - if self.should_stop(): - break - self.log(f"--- 开始第 {i + 1}/{count} 个账号 ---") - try: - email = "" - dev_token = "" - code = "" - mail_ok = False - max_mail_retry = 3 - for mail_try in range(1, max_mail_retry + 1): - self.log(f"[*] 1. 打开注册页 (尝试 {mail_try}/{max_mail_retry})") - open_signup_page( - log_callback=self.log, cancel_callback=self.should_stop - ) - self.log("[*] 2. 创建邮箱并提交") - email, dev_token = fill_email_and_submit( - log_callback=self.log, cancel_callback=self.should_stop - ) - self.log(f"[*] 邮箱: {email}") - self.log(f"[Debug] 邮箱credential(jwt): {dev_token}") - try: - with open( - os.path.join(os.path.dirname(__file__), "mail_credentials.txt"), - "a", - encoding="utf-8", - ) as f: - f.write(f"{email}\t{dev_token}\n") - except Exception: - pass - self.log("[*] 3. 拉取验证码") - try: - code = fill_code_and_submit( - email, - dev_token, - log_callback=self.log, - cancel_callback=self.should_stop, - ) - mail_ok = True - break - except Exception as mail_exc: - msg = str(mail_exc) - if ("未收到验证码" in msg or "验证码" in msg) and mail_try < max_mail_retry: - self.log(f"[!] 本邮箱未取到验证码，自动更换新邮箱重试: {msg}") - restart_browser(log_callback=self.log) - sleep_with_cancel(1, self.should_stop) - continue - raise - - if not mail_ok: - raise Exception("验证码阶段失败，已达到最大重试次数") - self.log(f"[*] 验证码: {code}") - self.log("[*] 4. 填写资料") - profile = fill_profile_and_submit( - log_callback=self.log, cancel_callback=self.should_stop - ) - self.log(f"[*] 资料已填: {profile.get('given_name')} {profile.get('family_name')}") - self.log("[*] 5. 等待 sso cookie") - sso = wait_for_sso_cookie( - log_callback=self.log, cancel_callback=self.should_stop - ) - if config.get("enable_nsfw", True): - self.log("[*] 6. 开启 NSFW") - nsfw_ok, nsfw_msg = enable_nsfw_for_token( - sso, log_callback=self.log - ) - if nsfw_ok: - self.log(f"[+] NSFW 开启成功: {nsfw_msg}") - else: - self.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") - self.results.append({"email": email, "sso": sso, "profile": profile}) - try: - line = f"{email}----{profile.get('password','')}----{sso}\n" - with open(self.accounts_output_file, "a", encoding="utf-8") as f: - f.write(line) - except Exception as file_exc: - self.log(f"[Debug] 保存账号文件失败: {file_exc}") - add_token_to_grok2api_pools(sso, email=email, log_callback=self.log) - maybe_export_cpa_xai_after_success( - email=email, - password=profile.get("password", ""), - sso=sso, - log_callback=self.log, - cancel_callback=self.should_stop, - ) - self.success_count += 1 - retry_count_for_slot = 0 - i += 1 - self.log(f"[+] 注册成功: {email}") - if ( - self.success_count > 0 - and self.success_count % MEMORY_CLEANUP_INTERVAL == 0 - and i < count - ): - cleanup_runtime_memory( - log_callback=self.log, - reason=f"已成功 {self.success_count} 个账号，执行定期清理", - ) - except RegistrationCancelled: - self.log("[!] 注册被用户停止") - break - except AccountRetryNeeded as exc: - retry_count_for_slot += 1 - if retry_count_for_slot <= max_slot_retry: - self.log( - f"[!] 当前账号流程卡住，重试第 {retry_count_for_slot}/{max_slot_retry} 次: {exc}" - ) - else: - self.fail_count += 1 - self.log( - f"[-] 当前账号已达到最大重试次数，跳过: {exc}" - ) - retry_count_for_slot = 0 - i += 1 - except Exception as exc: - self.fail_count += 1 - retry_count_for_slot = 0 - i += 1 - self.log(f"[-] 注册失败: {exc}") - finally: - self.update_stats() - if self.should_stop(): - break - if browser is None: - start_browser(log_callback=self.log) - else: - restart_browser(log_callback=self.log) - sleep_with_cancel(1, self.should_stop) + batch = run_registration_common( + count=count, + log_callback=self.log, + cancel_callback=self.should_stop, + accounts_output_file=self.accounts_output_file, + observer=observer, + ) + self.success_count = batch.success_count + self.fail_count = batch.fail_count except Exception as exc: - self.log(f"[!] 任务异常: {exc}") + log_exception("任务异常", exc, self.log) finally: - cleanup_runtime_memory(log_callback=self.log, reason="任务结束") self._set_running_ui(False) self.log("[*] 任务结束") + + class CliStopController: def __init__(self): self.stop_requested = False @@ -3631,149 +3688,34 @@ def cli_log(message): def run_registration_cli(count): controller = CliStopController() - success_count = 0 - fail_count = 0 - retry_count_for_slot = 0 - max_slot_retry = 3 accounts_output_file = os.path.join( os.path.dirname(__file__), f"accounts_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", ) cli_log(f"[*] 终端模式启动，目标数量: {count}") cli_log(f"[*] 成功账号将实时保存到: {accounts_output_file}") + last_stats = {"success": 0, "fail": 0} + def observer(batch, account, output): + last_stats["success"] = batch.success_count + last_stats["fail"] = batch.fail_count + cli_log(f"[*] 当前统计: 成功 {batch.success_count} | 失败 {batch.fail_count}") try: - start_browser(log_callback=cli_log) - cli_log("[*] 浏览器已启动") - i = 0 - while i < count: - if controller.should_stop(): - break - cli_log(f"--- 开始第 {i + 1}/{count} 个账号 ---") - try: - email = "" - dev_token = "" - code = "" - mail_ok = False - max_mail_retry = 3 - for mail_try in range(1, max_mail_retry + 1): - cli_log(f"[*] 1. 打开注册页 (尝试 {mail_try}/{max_mail_retry})") - open_signup_page( - log_callback=cli_log, cancel_callback=controller.should_stop - ) - cli_log("[*] 2. 创建邮箱并提交") - email, dev_token = fill_email_and_submit( - log_callback=cli_log, cancel_callback=controller.should_stop - ) - cli_log(f"[*] 邮箱: {email}") - cli_log(f"[Debug] 邮箱credential(jwt): {dev_token}") - try: - with open( - os.path.join(os.path.dirname(__file__), "mail_credentials.txt"), - "a", - encoding="utf-8", - ) as f: - f.write(f"{email}\t{dev_token}\n") - except Exception: - pass - cli_log("[*] 3. 拉取验证码") - try: - code = fill_code_and_submit( - email, - dev_token, - log_callback=cli_log, - cancel_callback=controller.should_stop, - ) - mail_ok = True - break - except Exception as mail_exc: - msg = str(mail_exc) - if ("未收到验证码" in msg or "验证码" in msg) and mail_try < max_mail_retry: - cli_log(f"[!] 本邮箱未取到验证码，自动更换新邮箱重试: {msg}") - restart_browser(log_callback=cli_log) - sleep_with_cancel(1, controller.should_stop) - continue - raise - - if not mail_ok: - raise Exception("验证码阶段失败，已达到最大重试次数") - cli_log(f"[*] 验证码: {code}") - cli_log("[*] 4. 填写资料") - profile = fill_profile_and_submit( - log_callback=cli_log, cancel_callback=controller.should_stop - ) - cli_log(f"[*] 资料已填: {profile.get('given_name')} {profile.get('family_name')}") - cli_log("[*] 5. 等待 sso cookie") - sso = wait_for_sso_cookie( - log_callback=cli_log, cancel_callback=controller.should_stop - ) - if config.get("enable_nsfw", True): - cli_log("[*] 6. 开启 NSFW") - nsfw_ok, nsfw_msg = enable_nsfw_for_token( - sso, log_callback=cli_log - ) - if nsfw_ok: - cli_log(f"[+] NSFW 开启成功: {nsfw_msg}") - else: - cli_log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") - try: - line = f"{email}----{profile.get('password','')}----{sso}\n" - with open(accounts_output_file, "a", encoding="utf-8") as f: - f.write(line) - except Exception as file_exc: - cli_log(f"[Debug] 保存账号文件失败: {file_exc}") - add_token_to_grok2api_pools(sso, email=email, log_callback=cli_log) - maybe_export_cpa_xai_after_success( - email=email, - password=profile.get("password", ""), - sso=sso, - log_callback=cli_log, - cancel_callback=controller.should_stop, - ) - success_count += 1 - retry_count_for_slot = 0 - i += 1 - cli_log(f"[+] 注册成功: {email}") - cli_log(f"[*] 当前统计: 成功 {success_count} | 失败 {fail_count}") - if success_count > 0 and success_count % MEMORY_CLEANUP_INTERVAL == 0 and i < count: - cleanup_runtime_memory( - log_callback=cli_log, - reason=f"已成功 {success_count} 个账号，执行定期清理", - ) - except RegistrationCancelled: - cli_log("[!] 注册被停止") - break - except AccountRetryNeeded as exc: - retry_count_for_slot += 1 - if retry_count_for_slot <= max_slot_retry: - cli_log( - f"[!] 当前账号流程卡住，重试第 {retry_count_for_slot}/{max_slot_retry} 次: {exc}" - ) - else: - fail_count += 1 - retry_count_for_slot = 0 - i += 1 - cli_log(f"[-] 当前账号已达到最大重试次数，跳过: {exc}") - except Exception as exc: - fail_count += 1 - retry_count_for_slot = 0 - i += 1 - cli_log(f"[-] 注册失败: {exc}") - finally: - if controller.should_stop(): - break - if browser is None: - start_browser(log_callback=cli_log) - else: - restart_browser(log_callback=cli_log) - sleep_with_cancel(1, controller.should_stop) + batch = run_registration_common( + count=count, + log_callback=cli_log, + cancel_callback=controller.should_stop, + accounts_output_file=accounts_output_file, + observer=observer, + ) + last_stats["success"] = batch.success_count + last_stats["fail"] = batch.fail_count except KeyboardInterrupt: controller.stop() cli_log("[!] 收到 Ctrl+C，正在停止并清理") except Exception as exc: - cli_log(f"[!] 任务异常: {exc}") + log_exception("任务异常", exc, cli_log) finally: - cleanup_runtime_memory(log_callback=cli_log, reason="任务结束") - cli_log(f"[*] 任务结束。成功 {success_count} | 失败 {fail_count}") + cli_log(f"[*] 任务结束。成功 {last_stats['success']} | 失败 {last_stats['fail']}") def main_cli(): diff --git a/registration_flow.py b/registration_flow.py index 18e3c07..3d31720 100644 --- a/registration_flow.py +++ b/registration_flow.py @@ -1,12 +1,6 @@ -# -*- coding: utf-8 -*- -"""Shared registration workflow used by both GUI and CLI.""" - +"""Shared registration workflow used by both GUI and CLI adapters.""" from dataclasses import dataclass, field -from typing import Any, Callable, Dict, List, Type - - -def _noop(*_args, **_kwargs): - return None +from typing import Any, Callable, Dict, Optional, Tuple @dataclass @@ -15,31 +9,26 @@ class RegistrationCallbacks: cancelled: Callable[[], bool] -@dataclass -class RegistrationObserver: - on_stats: Callable[[int, int], None] = _noop - on_result: Callable[["RegistrationResult", "OutputResult"], None] = _noop - - @dataclass class RegistrationOperations: - start_browser: Callable[[Callable[[str], None]], None] + start_browser: Callable[[], None] + restart_browser: Callable[[], None] browser_missing: Callable[[], bool] - restart_browser: Callable[[Callable[[str], None]], None] - cleanup_runtime_memory: Callable[[Callable[[str], None], str], None] - sleep_with_cancel: Callable[[float, Callable[[], bool]], None] - open_signup_page: Callable[[Callable[[str], None], Callable[[], bool]], None] - fill_email_and_submit: Callable[[Callable[[str], None], Callable[[], bool]], Any] - save_mail_credential: Callable[[str, str], Dict[str, Any]] - fill_code_and_submit: Callable[[str, str, Callable[[str], None], Callable[[], bool]], str] - fill_profile_and_submit: Callable[[Callable[[str], None], Callable[[], bool]], Dict[str, Any]] - wait_for_sso_cookie: Callable[[Callable[[str], None], Callable[[], bool]], str] - enable_nsfw_for_token: Callable[[str, Callable[[str], None]], Any] - save_account_result: Callable[["RegistrationResult", str], Dict[str, Any]] - add_token_pools: Callable[["RegistrationResult", Callable[[str], None]], Dict[str, Any]] - export_cpa: Callable[["RegistrationResult", Callable[[str], None], Callable[[], bool]], Dict[str, Any]] - cancelled_error: Type[BaseException] - retry_error: Type[BaseException] + open_signup_page: Callable[[], None] + fill_email_and_submit: Callable[[], Tuple[str, str]] + save_mail_credential: Callable[[str, str], bool] + fill_code_and_submit: Callable[[str, str], str] + fill_profile_and_submit: Callable[[], Dict[str, Any]] + wait_for_sso_cookie: Callable[[], str] + enable_nsfw: Callable[[str], Tuple[bool, str]] + persist_account_line: Callable[[str, str, str], None] + queue_unsaved_result: Callable[[Dict[str, Any], str], bool] + add_tokens: Callable[[str, str], Dict[str, Dict[str, Any]]] + export_cpa: Callable[[str, str, str], Dict[str, Any]] + cleanup: Callable[[str], None] + sleep: Callable[[float], None] + cancelled_exception: type + retry_exception: type @dataclass @@ -53,101 +42,69 @@ class RegistrationResult: retryable: bool = False -@dataclass -class OutputContext: - accounts_output_file: str - save_attempts: int = 3 - - @dataclass class OutputResult: registered: bool saved: bool - save_attempts: int = 0 + pending_saved: bool = False save_error: str = "" - token_pools: Dict[str, Any] = field(default_factory=dict) + pools: Dict[str, Dict[str, Any]] = field(default_factory=dict) cpa: Dict[str, Any] = field(default_factory=dict) @dataclass class BatchResult: - requested: int success_count: int = 0 fail_count: int = 0 + processed_count: int = 0 cancelled: bool = False - results: List[RegistrationResult] = field(default_factory=list) - outputs: List[OutputResult] = field(default_factory=list) - pending_outputs: List[RegistrationResult] = field(default_factory=list) + results: list = field(default_factory=list) -def register_one_account( - callbacks: RegistrationCallbacks, - operations: RegistrationOperations, - enable_nsfw: bool = True, - max_mail_retry: int = 3, -) -> RegistrationResult: +def register_one_account(callbacks, ops, enable_nsfw=True, max_mail_retry=3): email = "" dev_token = "" code = "" + mail_ok = False for mail_try in range(1, max_mail_retry + 1): + if callbacks.cancelled(): + raise ops.cancelled_exception() callbacks.log(f"[*] 1. 打开注册页 (尝试 {mail_try}/{max_mail_retry})") - operations.open_signup_page(callbacks.log, callbacks.cancelled) + ops.open_signup_page() callbacks.log("[*] 2. 创建邮箱并提交") - email, dev_token = operations.fill_email_and_submit( - callbacks.log, callbacks.cancelled - ) + email, dev_token = ops.fill_email_and_submit() callbacks.log(f"[*] 邮箱: {email}") callbacks.log(f"[Debug] 邮箱credential(jwt): {dev_token}") - credential_status = operations.save_mail_credential(email, dev_token) - if not credential_status.get("ok"): - callbacks.log( - "[!] 邮箱凭据保存失败，注册流程继续但已记录警告: " - + str(credential_status.get("error") or "unknown error") - ) + if not ops.save_mail_credential(email, dev_token): + callbacks.log("[!] 邮箱凭据保存失败，注册继续，但已明确记录该异常") callbacks.log("[*] 3. 拉取验证码") try: - code = operations.fill_code_and_submit( - email, dev_token, callbacks.log, callbacks.cancelled - ) + code = ops.fill_code_and_submit(email, dev_token) + mail_ok = True break - except operations.cancelled_error: - raise except Exception as exc: message = str(exc) - if ( - ("未收到验证码" in message or "验证码" in message) - and mail_try < max_mail_retry - ): - callbacks.log( - f"[!] 本邮箱未取到验证码，自动更换新邮箱重试: {message}" - ) - operations.restart_browser(callbacks.log) - operations.sleep_with_cancel(1, callbacks.cancelled) + if ("未收到验证码" in message or "验证码" in message) and mail_try < max_mail_retry: + callbacks.log(f"[!] 本邮箱未取到验证码，自动更换新邮箱重试: {message}") + ops.restart_browser() + ops.sleep(1) continue raise - else: + if not mail_ok: raise RuntimeError("验证码阶段失败，已达到最大重试次数") - callbacks.log(f"[*] 验证码: {code}") callbacks.log("[*] 4. 填写资料") - profile = operations.fill_profile_and_submit( - callbacks.log, callbacks.cancelled - ) - callbacks.log( - f"[*] 资料已填: {profile.get('given_name', '')} " - f"{profile.get('family_name', '')}" - ) + profile = ops.fill_profile_and_submit() + callbacks.log(f"[*] 资料已填: {profile.get('given_name')} {profile.get('family_name')}") callbacks.log("[*] 5. 等待 sso cookie") - sso = operations.wait_for_sso_cookie(callbacks.log, callbacks.cancelled) + sso = ops.wait_for_sso_cookie() if enable_nsfw: callbacks.log("[*] 6. 开启 NSFW") - nsfw_ok, nsfw_message = operations.enable_nsfw_for_token( - sso, callbacks.log - ) + nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) if nsfw_ok: - callbacks.log(f"[+] NSFW 开启成功: {nsfw_message}") + callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") else: - callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_message}") + callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") return RegistrationResult( ok=True, email=email, @@ -157,157 +114,103 @@ def register_one_account( ) -def persist_account_result( - result: RegistrationResult, - context: OutputContext, - callbacks: RegistrationCallbacks, - operations: RegistrationOperations, -) -> OutputResult: - attempts = max(int(context.save_attempts), 1) - save_error = "" - saved = False - used_attempts = 0 - for attempt in range(1, attempts + 1): - used_attempts = attempt - status = operations.save_account_result( - result, context.accounts_output_file +def persist_account_result(result, callbacks, ops): + try: + ops.persist_account_line(result.email, result.password, result.sso) + saved = True + save_error = "" + pending_saved = False + except Exception as exc: + saved = False + save_error = str(exc) + pending_saved = ops.queue_unsaved_result( + { + "email": result.email, + "password": result.password, + "sso": result.sso, + "profile": result.profile, + }, + save_error, ) - if status.get("ok"): - saved = True - break - save_error = str(status.get("error") or "unknown error") - callbacks.log( - f"[!] 保存账号结果失败 ({attempt}/{attempts}): {save_error}" - ) - if attempt < attempts: - operations.sleep_with_cancel( - min(0.5 * attempt, 1.5), callbacks.cancelled - ) - - if not saved: - return OutputResult( - registered=True, - saved=False, - save_attempts=used_attempts, - save_error=save_error, - ) - - token_pools = operations.add_token_pools(result, callbacks.log) - for target in ("local", "remote"): - target_result = token_pools.get(target) or {} - if target_result.get("enabled") and not target_result.get("ok"): - callbacks.log( - f"[!] grok2api {target} 入池失败，账号已安全保存: " - f"{target_result.get('error') or 'unknown error'}" - ) - - cpa = operations.export_cpa( - result, callbacks.log, callbacks.cancelled - ) + callbacks.log(f"[!] 账号已注册但主结果文件保存失败: {save_error}") + if pending_saved: + callbacks.log("[!] 未保存账号已写入 pending 队列，等待人工重试") + else: + callbacks.log("[!] pending 队列也写入失败，请立即复制当前账号信息") + pools = ops.add_tokens(result.sso, result.email) + for name, state in pools.items(): + if state.get("enabled") and not state.get("ok"): + callbacks.log(f"[!] grok2api {name} 入池失败: {state.get('error')}") + cpa = ops.export_cpa(result.email, result.password, result.sso) return OutputResult( registered=True, - saved=True, - save_attempts=used_attempts, - token_pools=token_pools, + saved=saved, + pending_saved=pending_saved, + save_error=save_error, + pools=pools, cpa=cpa, ) -def run_batch( - count: int, - callbacks: RegistrationCallbacks, - observer: RegistrationObserver, - operations: RegistrationOperations, - output_context: OutputContext, - enable_nsfw: bool = True, - cleanup_interval: int = 5, - max_slot_retry: int = 3, -) -> BatchResult: - batch = BatchResult(requested=count) +def run_batch(count, callbacks, observer, ops, enable_nsfw=True, cleanup_interval=5, + max_slot_retry=3, max_mail_retry=3): + result = BatchResult() retry_count_for_slot = 0 - index = 0 - operations.start_browser(callbacks.log) + ops.start_browser() callbacks.log("[*] 浏览器已启动") try: - while index < count: + while result.processed_count < count: if callbacks.cancelled(): - batch.cancelled = True + result.cancelled = True break - callbacks.log(f"--- 开始第 {index + 1}/{count} 个账号 ---") + callbacks.log(f"--- 开始第 {result.processed_count + 1}/{count} 个账号 ---") + account = None + output = None try: - registration = register_one_account( - callbacks, - operations, - enable_nsfw=enable_nsfw, + account = register_one_account( + callbacks, ops, enable_nsfw=enable_nsfw, + max_mail_retry=max_mail_retry, ) - output = persist_account_result( - registration, - output_context, - callbacks, - operations, - ) - batch.results.append(registration) - batch.outputs.append(output) - observer.on_result(registration, output) + output = persist_account_result(account, callbacks, ops) + result.results.append({"registration": account, "output": output}) retry_count_for_slot = 0 - index += 1 + result.processed_count += 1 if output.saved: - batch.success_count += 1 - callbacks.log( - f"[+] 注册并保存成功: {registration.email}" - ) - if ( - cleanup_interval > 0 - and batch.success_count % cleanup_interval == 0 - and index < count - ): - operations.cleanup_runtime_memory( - callbacks.log, - f"已成功 {batch.success_count} 个账号，执行定期清理", - ) + result.success_count += 1 + callbacks.log(f"[+] 注册并保存成功: {account.email}") else: - batch.fail_count += 1 - batch.pending_outputs.append(registration) - callbacks.log( - "[-] 账号已注册但结果未能持久化，未计入成功并加入" - f"待重试队列: {registration.email}: " - f"{output.save_error}" - ) - except operations.cancelled_error: - batch.cancelled = True + result.fail_count += 1 + callbacks.log(f"[-] 注册成功但持久化未完成: {account.email}") + if result.success_count > 0 and result.success_count % cleanup_interval == 0 and result.processed_count < count: + ops.cleanup(f"已成功 {result.success_count} 个账号，执行定期清理") + except ops.cancelled_exception: + result.cancelled = True callbacks.log("[!] 注册被停止") break - except operations.retry_error as exc: + except ops.retry_exception as exc: retry_count_for_slot += 1 if retry_count_for_slot <= max_slot_retry: - callbacks.log( - "[!] 当前账号流程卡住，重试第 " - f"{retry_count_for_slot}/{max_slot_retry} 次: {exc}" - ) + callbacks.log(f"[!] 当前账号流程卡住，重试第 {retry_count_for_slot}/{max_slot_retry} 次: {exc}") else: - batch.fail_count += 1 + result.fail_count += 1 + result.processed_count += 1 retry_count_for_slot = 0 - index += 1 - callbacks.log( - f"[-] 当前账号已达到最大重试次数，跳过: {exc}" - ) + callbacks.log(f"[-] 当前账号已达到最大重试次数，跳过: {exc}") except Exception as exc: - batch.fail_count += 1 + result.fail_count += 1 + result.processed_count += 1 retry_count_for_slot = 0 - index += 1 callbacks.log(f"[-] 注册失败: {exc}") finally: - observer.on_stats( - batch.success_count, batch.fail_count - ) + observer(result, account, output) if callbacks.cancelled(): - batch.cancelled = True + result.cancelled = True break - if operations.browser_missing(): - operations.start_browser(callbacks.log) + if ops.browser_missing(): + ops.start_browser() else: - operations.restart_browser(callbacks.log) - operations.sleep_with_cancel(1, callbacks.cancelled) + ops.restart_browser() + ops.sleep(1) finally: - operations.cleanup_runtime_memory(callbacks.log, "任务结束") - return batch + ops.cleanup("任务结束") + return result diff --git a/tools/apply_remaining_hardening.py b/tools/apply_remaining_hardening.py deleted file mode 100644 index 4a61220..0000000 --- a/tools/apply_remaining_hardening.py +++ /dev/null @@ -1,890 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -from pathlib import Path -import ast - -ROOT = Path(__file__).resolve().parents[1] -APP = ROOT / "grok_register_ttk.py" -FLOW = ROOT / "registration_flow.py" -OAUTH = ROOT / "cpa_xai" / "oauth_device.py" -BROWSER = ROOT / "cpa_xai" / "browser_confirm.py" -EXPORT = ROOT / "cpa_export.py" -MINT = ROOT / "cpa_xai" / "mint.py" - - -def read(path): - return path.read_text(encoding="utf-8-sig") - - -def write(path, text): - path.write_text(text, encoding="utf-8") - - -def replace_once(text, old, new, label): - count = text.count(old) - if count != 1: - raise RuntimeError(f"{label}: expected one match, got {count}") - return text.replace(old, new, 1) - - -def replace_between(text, start, end, new, label): - i = text.find(start) - if i < 0: - raise RuntimeError(f"{label}: start not found") - j = text.find(end, i + len(start)) - if j < 0: - raise RuntimeError(f"{label}: end not found") - return text[:i] + new + text[j:] - - -flow = r'''"""Shared registration workflow used by both GUI and CLI adapters.""" -from dataclasses import dataclass, field -from typing import Any, Callable, Dict, Optional, Tuple - - -@dataclass -class RegistrationCallbacks: - log: Callable[[str], None] - cancelled: Callable[[], bool] - - -@dataclass -class RegistrationOperations: - start_browser: Callable[[], None] - restart_browser: Callable[[], None] - browser_missing: Callable[[], bool] - open_signup_page: Callable[[], None] - fill_email_and_submit: Callable[[], Tuple[str, str]] - save_mail_credential: Callable[[str, str], bool] - fill_code_and_submit: Callable[[str, str], str] - fill_profile_and_submit: Callable[[], Dict[str, Any]] - wait_for_sso_cookie: Callable[[], str] - enable_nsfw: Callable[[str], Tuple[bool, str]] - persist_account_line: Callable[[str, str, str], None] - queue_unsaved_result: Callable[[Dict[str, Any], str], bool] - add_tokens: Callable[[str, str], Dict[str, Dict[str, Any]]] - export_cpa: Callable[[str, str, str], Dict[str, Any]] - cleanup: Callable[[str], None] - sleep: Callable[[float], None] - cancelled_exception: type - retry_exception: type - - -@dataclass -class RegistrationResult: - ok: bool - email: str = "" - password: str = "" - sso: str = "" - profile: Dict[str, Any] = field(default_factory=dict) - error: str = "" - retryable: bool = False - - -@dataclass -class OutputResult: - registered: bool - saved: bool - pending_saved: bool = False - save_error: str = "" - pools: Dict[str, Dict[str, Any]] = field(default_factory=dict) - cpa: Dict[str, Any] = field(default_factory=dict) - - -@dataclass -class BatchResult: - success_count: int = 0 - fail_count: int = 0 - processed_count: int = 0 - cancelled: bool = False - results: list = field(default_factory=list) - - -def register_one_account(callbacks, ops, enable_nsfw=True, max_mail_retry=3): - email = "" - dev_token = "" - code = "" - mail_ok = False - for mail_try in range(1, max_mail_retry + 1): - if callbacks.cancelled(): - raise ops.cancelled_exception() - callbacks.log(f"[*] 1. 打开注册页 (尝试 {mail_try}/{max_mail_retry})") - ops.open_signup_page() - callbacks.log("[*] 2. 创建邮箱并提交") - email, dev_token = ops.fill_email_and_submit() - callbacks.log(f"[*] 邮箱: {email}") - callbacks.log(f"[Debug] 邮箱credential(jwt): {dev_token}") - if not ops.save_mail_credential(email, dev_token): - callbacks.log("[!] 邮箱凭据保存失败，注册继续，但已明确记录该异常") - callbacks.log("[*] 3. 拉取验证码") - try: - code = ops.fill_code_and_submit(email, dev_token) - mail_ok = True - break - except Exception as exc: - message = str(exc) - if ("未收到验证码" in message or "验证码" in message) and mail_try < max_mail_retry: - callbacks.log(f"[!] 本邮箱未取到验证码，自动更换新邮箱重试: {message}") - ops.restart_browser() - ops.sleep(1) - continue - raise - if not mail_ok: - raise RuntimeError("验证码阶段失败，已达到最大重试次数") - callbacks.log(f"[*] 验证码: {code}") - callbacks.log("[*] 4. 填写资料") - profile = ops.fill_profile_and_submit() - callbacks.log(f"[*] 资料已填: {profile.get('given_name')} {profile.get('family_name')}") - callbacks.log("[*] 5. 等待 sso cookie") - sso = ops.wait_for_sso_cookie() - if enable_nsfw: - callbacks.log("[*] 6. 开启 NSFW") - nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) - if nsfw_ok: - callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") - else: - callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") - return RegistrationResult( - ok=True, - email=email, - password=str(profile.get("password") or ""), - sso=sso, - profile=profile, - ) - - -def persist_account_result(result, callbacks, ops): - try: - ops.persist_account_line(result.email, result.password, result.sso) - saved = True - save_error = "" - pending_saved = False - except Exception as exc: - saved = False - save_error = str(exc) - pending_saved = ops.queue_unsaved_result( - { - "email": result.email, - "password": result.password, - "sso": result.sso, - "profile": result.profile, - }, - save_error, - ) - callbacks.log(f"[!] 账号已注册但主结果文件保存失败: {save_error}") - if pending_saved: - callbacks.log("[!] 未保存账号已写入 pending 队列，等待人工重试") - else: - callbacks.log("[!] pending 队列也写入失败，请立即复制当前账号信息") - pools = ops.add_tokens(result.sso, result.email) - for name, state in pools.items(): - if state.get("enabled") and not state.get("ok"): - callbacks.log(f"[!] grok2api {name} 入池失败: {state.get('error')}") - cpa = ops.export_cpa(result.email, result.password, result.sso) - return OutputResult( - registered=True, - saved=saved, - pending_saved=pending_saved, - save_error=save_error, - pools=pools, - cpa=cpa, - ) - - -def run_batch(count, callbacks, observer, ops, enable_nsfw=True, cleanup_interval=5, - max_slot_retry=3, max_mail_retry=3): - result = BatchResult() - retry_count_for_slot = 0 - ops.start_browser() - callbacks.log("[*] 浏览器已启动") - try: - while result.processed_count < count: - if callbacks.cancelled(): - result.cancelled = True - break - callbacks.log(f"--- 开始第 {result.processed_count + 1}/{count} 个账号 ---") - account = None - output = None - try: - account = register_one_account( - callbacks, ops, enable_nsfw=enable_nsfw, - max_mail_retry=max_mail_retry, - ) - output = persist_account_result(account, callbacks, ops) - result.results.append({"registration": account, "output": output}) - retry_count_for_slot = 0 - result.processed_count += 1 - if output.saved: - result.success_count += 1 - callbacks.log(f"[+] 注册并保存成功: {account.email}") - else: - result.fail_count += 1 - callbacks.log(f"[-] 注册成功但持久化未完成: {account.email}") - if result.success_count > 0 and result.success_count % cleanup_interval == 0 and result.processed_count < count: - ops.cleanup(f"已成功 {result.success_count} 个账号，执行定期清理") - except ops.cancelled_exception: - result.cancelled = True - callbacks.log("[!] 注册被停止") - break - except ops.retry_exception as exc: - retry_count_for_slot += 1 - if retry_count_for_slot <= max_slot_retry: - callbacks.log(f"[!] 当前账号流程卡住，重试第 {retry_count_for_slot}/{max_slot_retry} 次: {exc}") - else: - result.fail_count += 1 - result.processed_count += 1 - retry_count_for_slot = 0 - callbacks.log(f"[-] 当前账号已达到最大重试次数，跳过: {exc}") - except Exception as exc: - result.fail_count += 1 - result.processed_count += 1 - retry_count_for_slot = 0 - callbacks.log(f"[-] 注册失败: {exc}") - finally: - observer(result, account, output) - if callbacks.cancelled(): - result.cancelled = True - break - if ops.browser_missing(): - ops.start_browser() - else: - ops.restart_browser() - ops.sleep(1) - finally: - ops.cleanup("任务结束") - return result -''' -write(FLOW, flow) -ast.parse(flow) - -app = read(APP) -app = replace_once(app, "import tempfile\n", "import tempfile\nimport traceback\n", "traceback import") -app = replace_once(app, ' "cpa_mint_cookie_inject": True,\n', ' "cpa_mint_cookie_inject": True,\n "cpa_oidc_request_timeout_sec": 15,\n "cpa_oidc_poll_timeout_sec": 15,\n "grok2api_allow_legacy_full_save": False,\n', "config defaults") - -config_block = r'''def _require_bool(cfg, key): - value = cfg.get(key) - if type(value) is not bool: - raise ConfigError(f"配置项 {key} 必须是布尔值 true/false") - return value - - -def _require_int(cfg, key, minimum, maximum): - value = cfg.get(key) - if type(value) is not int: - raise ConfigError(f"配置项 {key} 必须是整数") - if not minimum <= value <= maximum: - raise ConfigError(f"配置项 {key} 必须在 {minimum} 到 {maximum} 之间") - return value - - -def _require_string(cfg, key, path=False): - value = cfg.get(key) - if not isinstance(value, str): - raise ConfigError(f"配置项 {key} 必须是字符串") - value = value.strip() if key not in ("user_agent",) else value - if "\x00" in value: - raise ConfigError(f"配置项 {key} 包含非法空字符") - if path and value: - os.path.expanduser(value) - return value - - -def validate_config(raw): - if not isinstance(raw, dict): - raise ConfigError("config root must be a JSON object") - cfg = {**DEFAULT_CONFIG, **raw} - bool_keys = ( - "enable_nsfw", "grok2api_auto_add_local", "grok2api_auto_add_remote", - "grok2api_allow_legacy_full_save", "cpa_export_enabled", - "cpa_copy_to_hotload", "cpa_headless", "cpa_force_standalone", - "cpa_mint_cookie_inject", - ) - for key in bool_keys: - cfg[key] = _require_bool(cfg, key) - cfg["register_count"] = _require_int(cfg, "register_count", 1, 2500) - cfg["cpa_mint_timeout_sec"] = _require_int(cfg, "cpa_mint_timeout_sec", 30, 1800) - cfg["cpa_oidc_request_timeout_sec"] = _require_int(cfg, "cpa_oidc_request_timeout_sec", 3, 120) - cfg["cpa_oidc_poll_timeout_sec"] = _require_int(cfg, "cpa_oidc_poll_timeout_sec", 3, 120) - string_keys = tuple(key for key, value in DEFAULT_CONFIG.items() if isinstance(value, str)) - path_keys = {"grok2api_local_token_file", "api_reverse_tools", "cpa_auth_dir", "cpa_hotload_dir"} - for key in string_keys: - cfg[key] = _require_string(cfg, key, path=key in path_keys) - enums = { - "email_provider": {"duckmail", "yyds", "cloudflare", "cloudmail"}, - "cloudflare_auth_mode": {"query-key", "bearer", "x-api-key", "x-admin-auth", "none"}, - "grok2api_pool_name": {"ssoBasic", "ssoSuper"}, - } - for key, allowed in enums.items(): - value = cfg.get(key, DEFAULT_CONFIG.get(key, "")) - if value not in allowed: - raise ConfigError(f"配置项 {key} 的值无效: {value!r}; 允许值: {sorted(allowed)}") - cfg[key] = value - return cfg - - -def load_config(): - global config - if os.path.exists(CONFIG_FILE): - try: - with open(CONFIG_FILE, "r", encoding="utf-8") as f: - loaded = json.load(f) - config = validate_config(loaded) - except ConfigError: - raise - except Exception as exc: - raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc - else: - config = validate_config(DEFAULT_CONFIG.copy()) - return config - - -''' -app = replace_between(app, "def load_config():\n", "def save_config():\n", config_block, "config load/validation") -app = replace_once(app, "def save_config():\n config_dir", "def save_config():\n global config\n config = validate_config(config)\n config_dir", "validate before save") -app = replace_once(app, "\nload_config()\n\nEXTENSION_PATH", "\nEXTENSION_PATH", "remove import-time config load") - -insert_exceptions = r''' - -class RemoteTokenCompatibilityError(RuntimeError): - pass - - -class RemoteTokenRequestError(RuntimeError): - pass - - -def log_exception(context, exc, log_callback=None): - message = f"{context}: {exc.__class__.__name__}: {exc}" - if log_callback: - log_callback(f"[!] {message}") - else: - print(f"[!] {message}", file=sys.stderr) - return message -''' -app = replace_once(app, "class ConfigError(RuntimeError):\n pass\n", "class ConfigError(RuntimeError):\n pass\n" + insert_exceptions, "exception helpers") - -remote_block = r'''def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None): - token = _normalize_sso_token(raw_token) - if not token: - return False - base = str(config.get("grok2api_remote_base", "") or "").strip().rstrip("/") - app_key = str(config.get("grok2api_remote_app_key", "") or "").strip() - pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() - if not base or not app_key: - raise RemoteTokenRequestError("grok2api 远端未配置 base/app_key") - headers = {"Content-Type": "application/json"} - query = {"app_key": app_key} - remote_pool = {"ssoBasic": "basic", "ssoSuper": "super"}[pool_name] - api_bases = get_grok2api_remote_api_bases(base) - incompatible = [] - add_payload = {"tokens": [token], "pool": remote_pool, "tags": ["auto-register"]} - for api_base in api_bases: - endpoint = f"{api_base}/tokens/add" - try: - response = http_post(endpoint, headers=headers, params=query, json=add_payload, timeout=30) - except Exception as exc: - raise RemoteTokenRequestError(f"远端 /tokens/add 网络请求失败: {endpoint}: {exc}") from exc - status = int(getattr(response, "status_code", 0) or 0) - if 200 <= status < 300: - if log_callback: - log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({endpoint})") - return True - if status in (404, 405): - incompatible.append(f"{endpoint}: HTTP {status}") - continue - body = str(getattr(response, "text", "") or "")[:300] - raise RemoteTokenRequestError(f"远端 /tokens/add 请求失败，不允许全量回退: {endpoint}: HTTP {status}: {body}") - if not bool(config.get("grok2api_allow_legacy_full_save", False)): - raise RemoteTokenCompatibilityError( - "/tokens/add 不受支持，旧版全量保存默认禁用以避免并发覆盖: " + "; ".join(incompatible) - ) - current = None - fallback_base = None - etag = None - load_errors = [] - for api_base in api_bases or [base]: - endpoint = f"{api_base}/tokens" - try: - response = http_get(endpoint, headers=headers, params=query, timeout=20) - except Exception as exc: - raise RemoteTokenRequestError(f"旧版远端池读取网络失败: {endpoint}: {exc}") from exc - status = int(getattr(response, "status_code", 0) or 0) - if status != 200: - load_errors.append(f"{endpoint}: HTTP {status}") - continue - payload = response.json() - candidate = payload.get("tokens") if isinstance(payload, dict) and "tokens" in payload else payload - if not isinstance(candidate, dict): - load_errors.append(f"{endpoint}: unexpected payload") - continue - current = candidate - fallback_base = api_base - response_headers = getattr(response, "headers", {}) or {} - etag = response_headers.get("ETag") or response_headers.get("etag") - break - if current is None or fallback_base is None: - raise RemoteTokenRequestError("无法安全读取旧版远端 token 池: " + "; ".join(load_errors)) - pool = current.get(pool_name) - if pool is None: - pool = [] - elif not isinstance(pool, list): - raise RemoteTokenRequestError(f"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖") - existing = { - _normalize_sso_token(item if isinstance(item, str) else item.get("token", "")) - for item in pool if isinstance(item, (str, dict)) - } - if token not in existing: - pool.append({"token": token, "tags": ["auto-register"], "note": email}) - current[pool_name] = pool - save_headers = dict(headers) - if etag: - save_headers["If-Match"] = etag - elif log_callback: - log_callback("[!] 旧版远端接口未提供 ETag；已由显式配置允许，但仍不建议多实例并发") - endpoint = f"{fallback_base}/tokens" - try: - response = http_post(endpoint, headers=save_headers, params=query, json=current, timeout=30) - except Exception as exc: - raise RemoteTokenRequestError(f"旧版远端池保存网络失败: {endpoint}: {exc}") from exc - status = int(getattr(response, "status_code", 0) or 0) - if not 200 <= status < 300: - raise RemoteTokenRequestError(f"旧版远端池保存失败: {endpoint}: HTTP {status}") - if log_callback: - log_callback(f"[+] 已写入 grok2api 远端池（旧版兼容）: {pool_name} ({endpoint})") - return True - - -''' -app = replace_between(app, "def add_token_to_grok2api_remote_pool", "def add_token_to_grok2api_pools", remote_block, "remote token writer") - -pools_block = r'''def add_token_to_grok2api_pools(raw_token, email="", log_callback=None): - result = { - "local": {"enabled": bool(config.get("grok2api_auto_add_local", False)), "ok": None, "error": None}, - "remote": {"enabled": bool(config.get("grok2api_auto_add_remote", False)), "ok": None, "error": None}, - } - if result["local"]["enabled"]: - try: - result["local"]["ok"] = bool(add_token_to_grok2api_local_pool(raw_token, email=email, log_callback=log_callback)) - except Exception as exc: - result["local"]["ok"] = False - result["local"]["error"] = log_exception("写入 grok2api 本地池失败", exc, log_callback) - if result["remote"]["enabled"]: - try: - result["remote"]["ok"] = bool(add_token_to_grok2api_remote_pool(raw_token, email=email, log_callback=log_callback)) - except Exception as exc: - result["remote"]["ok"] = False - result["remote"]["error"] = log_exception("写入 grok2api 远端池失败", exc, log_callback) - return result - - -''' -app = replace_between(app, "def add_token_to_grok2api_pools", "def apply_browser_proxy_option", pools_block, "structured pool result") - -shared = r''' -def _save_mail_credential(email, credential, log_callback=None): - path = os.path.join(os.path.dirname(__file__), "mail_credentials.txt") - try: - with open(path, "a", encoding="utf-8") as handle: - handle.write(f"{email}\t{credential}\n") - handle.flush() - return True - except Exception as exc: - log_exception("保存邮箱凭据失败", exc, log_callback) - return False - - -def _append_account_line(path, email, password, sso): - with open(path, "a", encoding="utf-8") as handle: - handle.write(f"{email}----{password}----{sso}\n") - handle.flush() - os.fsync(handle.fileno()) - - -def _queue_unsaved_account(path, payload, error, log_callback=None): - pending_path = path + ".pending.jsonl" - record = dict(payload) - record["save_error"] = str(error) - record["queued_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() - try: - with open(pending_path, "a", encoding="utf-8") as handle: - handle.write(json.dumps(record, ensure_ascii=False) + "\n") - handle.flush() - os.fsync(handle.fileno()) - try: - os.chmod(pending_path, 0o600) - except Exception: - pass - return True - except Exception as exc: - log_exception("写入账号 pending 队列失败", exc, log_callback) - return False - - -def run_registration_common(count, log_callback, cancel_callback, accounts_output_file, observer): - from registration_flow import RegistrationCallbacks, RegistrationOperations, run_batch - callbacks = RegistrationCallbacks(log=log_callback, cancelled=cancel_callback) - operations = RegistrationOperations( - start_browser=lambda: start_browser(log_callback=log_callback), - restart_browser=lambda: restart_browser(log_callback=log_callback), - browser_missing=lambda: browser is None, - open_signup_page=lambda: open_signup_page(log_callback=log_callback, cancel_callback=cancel_callback), - fill_email_and_submit=lambda: fill_email_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), - save_mail_credential=lambda email, token: _save_mail_credential(email, token, log_callback), - fill_code_and_submit=lambda email, token: fill_code_and_submit(email, token, log_callback=log_callback, cancel_callback=cancel_callback), - fill_profile_and_submit=lambda: fill_profile_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), - wait_for_sso_cookie=lambda: wait_for_sso_cookie(log_callback=log_callback, cancel_callback=cancel_callback), - enable_nsfw=lambda sso: enable_nsfw_for_token(sso, log_callback=log_callback), - persist_account_line=lambda email, password, sso: _append_account_line(accounts_output_file, email, password, sso), - queue_unsaved_result=lambda payload, error: _queue_unsaved_account(accounts_output_file, payload, error, log_callback), - add_tokens=lambda sso, email: add_token_to_grok2api_pools(sso, email=email, log_callback=log_callback), - export_cpa=lambda email, password, sso: maybe_export_cpa_xai_after_success( - email=email, password=password, sso=sso, - log_callback=log_callback, cancel_callback=cancel_callback, - ), - cleanup=lambda reason: cleanup_runtime_memory(log_callback=log_callback, reason=reason), - sleep=lambda seconds: sleep_with_cancel(seconds, cancel_callback), - cancelled_exception=RegistrationCancelled, - retry_exception=AccountRetryNeeded, - ) - return run_batch( - count=count, - callbacks=callbacks, - observer=observer, - ops=operations, - enable_nsfw=bool(config.get("enable_nsfw", True)), - cleanup_interval=MEMORY_CLEANUP_INTERVAL, - max_slot_retry=3, - max_mail_retry=3, - ) - - -''' -app = replace_once(app, "class GrokRegisterGUI:\n", shared + "class GrokRegisterGUI:\n", "shared registration adapter") -app = replace_once(app, " self._ui_thread_id = threading.get_ident()\n self.accounts_output_file = \"\"\n self.setup_ui()\n", " self.accounts_output_file = \"\"\n self.setup_ui()\n self.root.after(50, self.process_ui_queue)\n", "GUI queue startup") -app = replace_once(app, " def setup_ui(self):\n load_config()\n", " def setup_ui(self):\n load_config()\n", "GUI load remains reachable") - -queue_methods = r''' def process_ui_queue(self): - try: - while True: - event = self.ui_queue.get_nowait() - kind = event[0] - if kind == "log": - line = event[1] - self.log_text.insert(tk.END, f"{line}\n") - self.log_text.see(tk.END) - elif kind == "stats": - self.stats_var.set(f"成功: {event[1]} | 失败: {event[2]}") - elif kind == "running": - running = bool(event[1]) - self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL) - self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED) - self.status_var.set("运行中..." if running else "就绪") - self.status_label.config(foreground="blue" if running else "green") - elif kind == "error": - messagebox.showerror(event[1], event[2]) - except queue.Empty: - pass - except Exception as exc: - print(f"[!] UI 队列处理失败: {exc}", file=sys.stderr) - finally: - try: - self.root.after(50, self.process_ui_queue) - except Exception: - pass - - def log(self, message): - timestamp = datetime.datetime.now().strftime("%H:%M:%S") - line = f"[{timestamp}] {message}" - print(line, flush=True) - self.ui_queue.put(("log", line)) - - def clear_log(self): - self.log_text.delete(1.0, tk.END) - - def update_stats(self): - self.ui_queue.put(("stats", self.success_count, self.fail_count)) - - def _set_running_ui(self, running): - self.is_running = bool(running) - self.ui_queue.put(("running", self.is_running)) - - -''' -app = replace_between(app, " def _call_ui", " def should_stop", queue_methods, "GUI main-thread queue") - -new_gui_run = r''' def run_registration(self, count): - def observer(batch, account, output): - self.success_count = batch.success_count - self.fail_count = batch.fail_count - if account is not None: - self.results.append({"email": account.email, "sso": account.sso, "profile": account.profile, "output": output}) - self.update_stats() - try: - batch = run_registration_common( - count=count, - log_callback=self.log, - cancel_callback=self.should_stop, - accounts_output_file=self.accounts_output_file, - observer=observer, - ) - self.success_count = batch.success_count - self.fail_count = batch.fail_count - except Exception as exc: - log_exception("任务异常", exc, self.log) - finally: - self._set_running_ui(False) - self.log("[*] 任务结束") - - -''' -app = replace_between(app, " def run_registration(self, count):", "\n\nclass CliStopController:", new_gui_run, "GUI shared batch") - -new_cli = r'''def run_registration_cli(count): - controller = CliStopController() - accounts_output_file = os.path.join( - os.path.dirname(__file__), - f"accounts_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", - ) - cli_log(f"[*] 终端模式启动，目标数量: {count}") - cli_log(f"[*] 成功账号将实时保存到: {accounts_output_file}") - last_stats = {"success": 0, "fail": 0} - def observer(batch, account, output): - last_stats["success"] = batch.success_count - last_stats["fail"] = batch.fail_count - cli_log(f"[*] 当前统计: 成功 {batch.success_count} | 失败 {batch.fail_count}") - try: - batch = run_registration_common( - count=count, - log_callback=cli_log, - cancel_callback=controller.should_stop, - accounts_output_file=accounts_output_file, - observer=observer, - ) - last_stats["success"] = batch.success_count - last_stats["fail"] = batch.fail_count - except KeyboardInterrupt: - controller.stop() - cli_log("[!] 收到 Ctrl+C，正在停止并清理") - except Exception as exc: - log_exception("任务异常", exc, cli_log) - finally: - cli_log(f"[*] 任务结束。成功 {last_stats['success']} | 失败 {last_stats['fail']}") - - -''' -app = replace_between(app, "def run_registration_cli(count):", "def main_cli():", new_cli, "CLI shared batch") -ast.parse(app) -write(APP, app) - -oauth = read(OAUTH) -oauth = replace_once(oauth, "def discover(proxy=None, timeout=30.0):", "def discover(proxy=None, timeout=30.0, cancel=None, retries=2):", "discover signature") -old_discover_body_start = "def discover(proxy=None, timeout=30.0, cancel=None, retries=2):\n" -new_discover = r'''def discover(proxy=None, timeout=30.0, cancel=None, retries=2): - request = urllib.request.Request( - DISCOVERY_URL, - method="GET", - headers={"Accept": "application/json", "User-Agent": "grok-register-cpa/1.0"}, - ) - last_error = None - for attempt in range(max(int(retries), 0) + 1): - _check_cancel(cancel) - opener = _build_opener(proxy) - try: - with opener.open(request, timeout=float(timeout)) as response: - body = response.read().decode("utf-8", errors="replace") - status = int(getattr(response, "status", 200) or 200) - _check_cancel(cancel) - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8", errors="replace") - raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (exc.code, body)) - except Exception as exc: - last_error = exc - if not _is_transient_net_error(exc) or attempt >= int(retries): - raise OAuthDeviceError("xAI discovery request failed: %s" % exc) - _sleep_with_cancel(1.0 * (attempt + 1), cancel) - continue - if status != 200: - raise OAuthDeviceError("xAI discovery failed HTTP %s: %s" % (status, body)) - try: - payload = json.loads(body) - except Exception as exc: - raise OAuthDeviceError("xAI discovery parse failed: %s" % exc) - return { - "device_authorization_endpoint": _validate_endpoint( - payload.get("device_authorization_endpoint"), "device_authorization_endpoint" - ), - "token_endpoint": _validate_endpoint(payload.get("token_endpoint"), "token_endpoint"), - } - raise OAuthDeviceError("xAI discovery failed: %s" % last_error) - - -''' -oauth = replace_between(oauth, old_discover_body_start, "def _is_transient_net_error", new_discover, "cancellable discovery") -helper = r'''def _check_cancel(cancel): - if cancel and cancel(): - raise OAuthDeviceError("cancelled") - - -def _sleep_with_cancel(seconds, cancel=None): - deadline = time.time() + max(float(seconds), 0.0) - while time.time() < deadline: - _check_cancel(cancel) - time.sleep(min(0.2, max(deadline - time.time(), 0.0))) - _check_cancel(cancel) - - -''' -oauth = replace_once(oauth, "def _is_transient_net_error", helper + "def _is_transient_net_error", "cancel helpers") -oauth = replace_once(oauth, "def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5):", "def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5, cancel=None):", "post cancel signature") -oauth = replace_once(oauth, " for attempt in range(max(int(retries), 0) + 1):\n opener = _build_opener(proxy)\n", " for attempt in range(max(int(retries), 0) + 1):\n _check_cancel(cancel)\n opener = _build_opener(proxy)\n", "post precheck") -oauth = replace_once(oauth, " status = int(getattr(response, \"status\", 200) or 200)\n", " status = int(getattr(response, \"status\", 200) or 200)\n _check_cancel(cancel)\n", "post postcheck") -oauth = replace_once(oauth, " time.sleep(float(retry_sleep) * (attempt + 1))\n", " _sleep_with_cancel(float(retry_sleep) * (attempt + 1), cancel)\n", "post cancel sleep") -oauth = replace_between(oauth, "def request_device_code", "def _sleep_with_cancel", r'''def request_device_code(client_id=CLIENT_ID, scope=SCOPE, timeout=15.0, proxy=None, cancel=None, retries=2): - discovery = discover(proxy=proxy, timeout=timeout, cancel=cancel, retries=retries) - _check_cancel(cancel) - device_endpoint = discovery["device_authorization_endpoint"] - token_endpoint = discovery["token_endpoint"] - status, payload = _post_form( - device_endpoint, - {"client_id": client_id, "scope": scope}, - timeout=timeout, - proxy=proxy, - retries=retries, - retry_sleep=1.0, - cancel=cancel, - ) - _check_cancel(cancel) - if status != 200 or not isinstance(payload, dict): - raise OAuthDeviceError("device code request failed HTTP %s: %r" % (status, payload)) - device_code = str(payload.get("device_code") or "").strip() - user_code = str(payload.get("user_code") or "").strip() - if not device_code or not user_code: - raise OAuthDeviceError("device code response missing fields: %r" % payload) - verification_uri = str(payload.get("verification_uri") or "https://accounts.x.ai/oauth2/device").strip() - verification_uri_complete = str( - payload.get("verification_uri_complete") or ("%s?user_code=%s" % (verification_uri, user_code)) - ).strip() - return DeviceCodeSession( - device_code=device_code, - user_code=user_code, - verification_uri=verification_uri, - verification_uri_complete=verification_uri_complete, - expires_in=int(payload.get("expires_in") or 1800), - interval=max(int(payload.get("interval") or 5), 1), - token_endpoint=token_endpoint, - raw=payload, - ) - - -''', "cancellable device request") -# Remove duplicate old helper left by boundary endpoint. -first = oauth.find("def _sleep_with_cancel") -second = oauth.find("def _sleep_with_cancel", first + 1) -if second >= 0: - end = oauth.find("def poll_device_token", second) - oauth = oauth[:second] + oauth[end:] -oauth = oauth.replace("timeout=min(float(timeout), 5.0)", "timeout=float(timeout)") -oauth = oauth.replace("retry_sleep=1.0,\n )", "retry_sleep=1.0,\n cancel=cancel,\n )", 1) -ast.parse(oauth) -write(OAUTH, oauth) - -browser = read(BROWSER) -old_create = r''' resolved = resolve_proxy(proxy) - proxy_bridge = None - chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger) - try: - if chrome_proxy: - options.set_argument("--proxy-server=%s" % chrome_proxy) - logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy)) - else: - logger("browser proxy=(none)") - - browser = Chromium(options) - if proxy_bridge is not None: - try: - setattr(browser, "_cpa_proxy_bridge", proxy_bridge) - except Exception: - pass - _register_mint_browser(browser) - page = browser.latest_tab - logger("standalone chromium started") - return browser, page - except Exception: - if proxy_bridge is not None: - try: - proxy_bridge.stop() - except Exception: - pass - raise -''' -new_create = r''' resolved = resolve_proxy(proxy) - proxy_bridge = None - browser = None - chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger) - try: - if chrome_proxy: - options.set_argument("--proxy-server=%s" % chrome_proxy) - logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy)) - else: - logger("browser proxy=(none)") - browser = Chromium(options) - if proxy_bridge is not None: - setattr(browser, "_cpa_proxy_bridge", proxy_bridge) - page = browser.latest_tab - _register_mint_browser(browser) - logger("standalone chromium started") - return browser, page - except Exception: - if browser is not None: - close_standalone(browser) - elif proxy_bridge is not None: - try: - proxy_bridge.stop() - except Exception: - pass - raise -''' -browser = replace_once(browser, old_create, new_create, "CPA browser creation cleanup") -browser = replace_once(browser, " recycle_every: int = 15,\n):", " recycle_every: int = 15,\n request_timeout_sec: float = 15.0,\n poll_timeout_sec: float = 15.0,\n):", "mint timeout params") -old_retry = r''' last_error = None - session = None - for attempt in range(1, 4): - try: - session = request_device_code(proxy=resolved or None) - last_error = None - break - except Exception as exc: - last_error = exc - logger("request_device_code attempt %s/3 failed: %s" % (attempt, exc)) - _sleep(1.5 * attempt) - if session is None: - raise last_error or RuntimeError("request_device_code failed") -''' -new_retry = r''' session = request_device_code( - proxy=resolved or None, - timeout=float(request_timeout_sec), - cancel=cancel, - retries=2, - ) -''' -browser = replace_once(browser, old_retry, new_retry, "single OIDC retry strategy") -browser = replace_once(browser, " proxy=resolved or None,\n )", " proxy=resolved or None,\n timeout=float(poll_timeout_sec),\n )", "poll configurable timeout") -ast.parse(browser) -write(BROWSER, browser) - -export = read(EXPORT) -export = replace_once(export, " timeout = float(cfg.get(\"cpa_mint_timeout_sec\") or 300)\n", " timeout = float(cfg.get(\"cpa_mint_timeout_sec\") or 300)\n request_timeout = float(cfg.get(\"cpa_oidc_request_timeout_sec\") or 15)\n poll_timeout = float(cfg.get(\"cpa_oidc_poll_timeout_sec\") or 15)\n", "export timeout config") -export = replace_once(export, " cancel=cancel_callback,\n )", " cancel=cancel_callback,\n request_timeout_sec=request_timeout,\n poll_timeout_sec=poll_timeout,\n )", "export timeout pass") -export = replace_once(export, " except Exception:\n pass\n return result\n", " except Exception as exc:\n log(\"[cpa] failed to persist failure record: %s\" % exc)\n return result\n", "CPA failure log") -ast.parse(export) -write(EXPORT, export) - -mint = read(MINT) -mint = replace_once(mint, " cancel=None,\n):", " cancel=None,\n request_timeout_sec=15.0,\n poll_timeout_sec=15.0,\n):", "mint API timeouts") -mint = replace_once(mint, " recycle_every=int(recycle_every or 0),\n )", " recycle_every=int(recycle_every or 0),\n request_timeout_sec=float(request_timeout_sec),\n poll_timeout_sec=float(poll_timeout_sec),\n )", "mint pass timeouts") -ast.parse(mint) -write(MINT, mint) - -for path in (APP, FLOW, OAUTH, BROWSER, EXPORT, MINT): - ast.parse(read(path)) -print("remaining hardening migration applied") diff --git a/tools/fix_remaining_patch_oidc_scope.py b/tools/fix_remaining_patch_oidc_scope.py deleted file mode 100644 index 730439d..0000000 --- a/tools/fix_remaining_patch_oidc_scope.py +++ /dev/null @@ -1,23 +0,0 @@ -#!/usr/bin/env python3 -from pathlib import Path - -script = Path(__file__).resolve().with_name("apply_remaining_hardening.py") -text = script.read_text(encoding="utf-8") -old = '''oauth = replace_once(oauth, " status = int(getattr(response, \\\"status\\\", 200) or 200)\\n", " status = int(getattr(response, \\\"status\\\", 200) or 200)\\n _check_cancel(cancel)\\n", "post postcheck")''' -new = '''post_start = oauth.find("def _post_form") -post_end = oauth.find("def request_device_code", post_start) -if post_start < 0 or post_end < 0: - raise RuntimeError("post postcheck: _post_form boundaries not found") -post_block = oauth[post_start:post_end] -post_block = replace_once( - post_block, - " status = int(getattr(response, \\\"status\\\", 200) or 200)\\n", - " status = int(getattr(response, \\\"status\\\", 200) or 200)\\n _check_cancel(cancel)\\n", - "post postcheck", -) -oauth = oauth[:post_start] + post_block + oauth[post_end:]''' -count = text.count(old) -if count != 1: - raise RuntimeError(f"OIDC postcheck patch anchor expected once, got {count}") -script.write_text(text.replace(old, new, 1), encoding="utf-8") -print("OIDC postcheck migration edit scoped to _post_form")