From 286a0f17346d54f1607b0a9802daf25140491a61 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:43:54 +0000 Subject: [PATCH] fix: harden token pools and CPA export --- .github/workflows/temp-audit-fixes.yml | 34 -- .gitignore | 2 + README.md | 7 +- cpa_xai/browser_confirm.py | 63 ++- cpa_xai/oauth_device.py | 20 +- cpa_xai/schema.py | 2 + cpa_xai/writer.py | 13 +- grok_register_ttk.py | 169 +++++-- requirements.txt | 2 +- tools/apply_audit_fixes.py | 643 ------------------------- tools/fix_audit_patch_escapes.py | 22 - 11 files changed, 210 insertions(+), 767 deletions(-) delete mode 100644 .github/workflows/temp-audit-fixes.yml delete mode 100644 tools/apply_audit_fixes.py delete mode 100644 tools/fix_audit_patch_escapes.py diff --git a/.github/workflows/temp-audit-fixes.yml b/.github/workflows/temp-audit-fixes.yml deleted file mode 100644 index cacd5d6..0000000 --- a/.github/workflows/temp-audit-fixes.yml +++ /dev/null @@ -1,34 +0,0 @@ -name: Temporary audit fixes - -on: - workflow_dispatch: - push: - branches: [main] - issues: - types: [edited, opened, reopened] - -permissions: - contents: write - -jobs: - apply: - if: github.event_name == 'workflow_dispatch' || github.event_name == 'push' || github.event.issue.number == 7 - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - name: Repair audit patch escapes - run: python tools/fix_audit_patch_escapes.py - - name: Apply audit fixes - run: python tools/apply_audit_fixes.py - - name: Commit and cleanup - 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_audit_fixes.py tools/fix_audit_patch_escapes.py .github/workflows/temp-audit-fixes.yml - git add -A - if git diff --cached --quiet; then - echo "No changes to commit" - else - git commit -m "fix: harden token pools and CPA export" - git push - fi diff --git a/.gitignore b/.gitignore index e9f0137..cbb962c 100755 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,5 @@ autoPortData/ cpa_auths/ xai-*.json cpa_auth_failed.txt +screenshots/ +*.png diff --git a/README.md b/README.md index de9572e..e637143 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,7 @@ GUI 模式会打开 Tkinter 窗口，适合手动调整配置和观察日志。 - `mail_credentials.txt`：临时邮箱凭证。 - `cpa_auths/xai-*.json`：可选导出的 CPA xAI OAuth 凭证。 - `cpa_auths/cpa_auth_failed.txt`：OIDC 导出失败记录。 +- `screenshots/`：CPA/OIDC 浏览器失败调试截图，已被 `.gitignore` 忽略。 - `*.log`：可选日志文件。 这些文件包含敏感信息，已被 `.gitignore` 忽略。 @@ -321,10 +322,14 @@ GUI 数量控件可能有上限。CLI 模式直接读取 `config.json` 中的 `r ```text . -├── grok_register_ttk.py # 主程序 +├── grok_register_ttk.py # 主程序（GUI / CLI） +├── cpa_export.py # 注册成功后的 CPA/OIDC 导出入口 +├── cpa_xai/ # xAI Device Auth、浏览器授权和凭证写入模块 ├── cf_mail_debug.py # Cloudflare 邮箱调试工具 ├── config.example.json # 配置示例 ├── requirements.txt # Python 依赖 +├── tests/ # 现有测试用例 +├── assets/ # README 资源 └── README.md ``` diff --git a/cpa_xai/browser_confirm.py b/cpa_xai/browser_confirm.py index 5d563f3..f0ddaca 100644 --- a/cpa_xai/browser_confirm.py +++ b/cpa_xai/browser_confirm.py @@ -189,10 +189,11 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False, pass break - from .proxyutil import proxy_for_chromium, proxy_log_label, resolve_proxy + from .proxyutil import prepare_chromium_proxy, proxy_log_label, resolve_proxy resolved = resolve_proxy(proxy) - chrome_proxy = proxy_for_chromium(resolved) + proxy_bridge = None + chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger) if chrome_proxy: options.set_argument("--proxy-server=%s" % chrome_proxy) logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy)) @@ -200,19 +201,50 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False, 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 def close_standalone(browser: Any) -> None: + if browser is None: + return + _unregister_mint_browser(browser) + bridge = getattr(browser, "_cpa_proxy_bridge", None) try: browser.quit() except Exception: pass + if bridge is not None: + try: + bridge.stop() + except Exception: + pass _mint_tls = threading.local() +_mint_registry_lock = threading.Lock() +_mint_registry = set() + + +def _register_mint_browser(browser: Any) -> None: + if browser is None: + return + with _mint_registry_lock: + _mint_registry.add(browser) + + +def _unregister_mint_browser(browser: Any) -> None: + if browser is None: + return + with _mint_registry_lock: + _mint_registry.discard(browser) def _mint_tls_get(): @@ -417,8 +449,9 @@ def release_mint_browser(owned: bool, success: bool, log: Optional[LogFn] = None def shutdown_mint_browsers() -> None: state = _mint_tls_get() - browser = state.get("browser") - if browser is not None: + with _mint_registry_lock: + browsers = list(_mint_registry) + for browser in browsers: try: close_standalone(browser) except Exception: @@ -824,7 +857,7 @@ def mint_with_browser( session = request_device_code(proxy=resolved or None) last_error = None break - except BaseException as exc: + except Exception as exc: last_error = exc logger("request_device_code attempt %s/3 failed: %s" % (attempt, exc)) _sleep(1.5 * attempt) @@ -852,22 +885,28 @@ def mint_with_browser( token_box = {} error_box = {} + def combined_cancel(): + return stop_event.is_set() or bool(cancel and cancel()) + def _poll() -> None: try: - time.sleep(2) + for _ in range(20): + if combined_cancel(): + raise OAuthDeviceError("cancelled") + time.sleep(0.1) result = poll_device_token( session.device_code, token_endpoint=session.token_endpoint, interval=max(session.interval, 5), expires_in=min(session.expires_in, int(browser_timeout_sec) + 60), log=logger, - cancel=cancel, + cancel=combined_cancel, proxy=resolved or None, ) token_box["token"] = result stop_event.set() logger("token poll SUCCESS — stop_event set") - except BaseException as exc: + except Exception as exc: error_box["err"] = exc stop_event.set() @@ -897,8 +936,16 @@ def mint_with_browser( ) if hard: stop_event.set() + thread.join(timeout=5) + if thread.is_alive(): + logger("token poll thread did not stop within 5s after browser failure") raise thread.join(timeout=max(browser_timeout_sec, 60) + 30) + if thread.is_alive(): + stop_event.set() + thread.join(timeout=5) + if thread.is_alive(): + raise OAuthDeviceError("token poll thread did not stop after timeout") if "token" in token_box: token_result = token_box["token"] success = True diff --git a/cpa_xai/oauth_device.py b/cpa_xai/oauth_device.py index 77f3b05..00ccb46 100644 --- a/cpa_xai/oauth_device.py +++ b/cpa_xai/oauth_device.py @@ -167,7 +167,7 @@ def _post_form(url, form, timeout=30.0, proxy=None, retries=0, retry_sleep=1.5): except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") status = int(exc.code) - except BaseException as exc: + except Exception as exc: last_error = exc if not _is_transient_net_error(exc) or attempt >= int(retries): raise @@ -216,6 +216,14 @@ 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, @@ -249,7 +257,7 @@ def poll_device_token( retry_sleep=1.0, ) net_streak = 0 - except BaseException as exc: + except Exception as exc: if not _is_transient_net_error(exc): raise net_streak += 1 @@ -257,7 +265,7 @@ def poll_device_token( logger("oauth poll network blip (%s/%s): %s — retry in %ss" % (net_streak, max_net_streak, exc, wait_seconds)) if net_streak >= max_net_streak: raise OAuthDeviceError("device auth aborted after %s network errors: %s" % (net_streak, exc)) - time.sleep(wait_seconds) + _sleep_with_cancel(wait_seconds, cancel) continue if status == 200 and isinstance(payload, dict) and payload.get("access_token"): access_token = str(payload.get("access_token") or "").strip() @@ -281,7 +289,7 @@ def poll_device_token( if error_code == "slow_down": sleep_seconds = min(sleep_seconds + 5, 30) logger("oauth poll: %s (sleep %ss)" % (error_code, sleep_seconds)) - time.sleep(sleep_seconds) + _sleep_with_cancel(sleep_seconds, cancel) continue if error_code in ("expired_token", "access_denied"): raise OAuthDeviceError("device auth failed: %s: %s" % (error_code, error_description)) @@ -293,8 +301,8 @@ def poll_device_token( logger("oauth poll soft HTTP %s: %r — retry in %ss" % (status, payload, wait_seconds)) if net_streak >= max_net_streak: raise OAuthDeviceError("device auth aborted after repeated soft HTTP failures status=%s" % status) - time.sleep(wait_seconds) + _sleep_with_cancel(wait_seconds, cancel) continue logger("oauth poll unexpected HTTP %s: %r" % (status, payload)) - time.sleep(sleep_seconds) + _sleep_with_cancel(sleep_seconds, cancel) raise OAuthDeviceError("device auth timed out waiting for user approval") diff --git a/cpa_xai/schema.py b/cpa_xai/schema.py index d4cf736..069580c 100644 --- a/cpa_xai/schema.py +++ b/cpa_xai/schema.py @@ -87,6 +87,8 @@ def build_cpa_xai_auth( expired = "" if exp: expired = datetime.fromtimestamp(exp, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + elif expires_in: + expired = datetime.fromtimestamp(time.time() + int(expires_in or 0), tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") payload = { "type": "xai", "access_token": access, diff --git a/cpa_xai/writer.py b/cpa_xai/writer.py index 1f3598b..8fd5224 100644 --- a/cpa_xai/writer.py +++ b/cpa_xai/writer.py @@ -8,13 +8,24 @@ from pathlib import Path from .schema import credential_file_name +def _is_relative_to(path, root): + try: + path.relative_to(root) + return True + except ValueError: + return False + + def write_cpa_xai_auth(auth_dir, payload, filename=None): root = Path(auth_dir).expanduser().resolve() root.mkdir(parents=True, exist_ok=True) target_name = filename or credential_file_name(payload.get("email", ""), payload.get("sub", "")) + target_name = Path(str(target_name)).name if not str(target_name).endswith(".json"): target_name = str(target_name) + ".json" - destination = root / str(target_name) + destination = (root / str(target_name)).resolve() + if not _is_relative_to(destination, root): + raise ValueError("CPA auth filename must stay inside auth_dir") data = json.dumps(payload, indent=2, ensure_ascii=False) + "\n" file_descriptor, temp_name = tempfile.mkstemp(prefix=".xai-", suffix=".tmp", dir=str(root)) try: diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 85fc3df..e6e06f4 100755 --- a/grok_register_ttk.py +++ b/grok_register_ttk.py @@ -1,12 +1,22 @@ -#!/usr/bin/env python +#!/usr/bin/env python # -*- coding: utf-8 -*- """ Grok 注册机 - TTK GUI 版本 整合 DrissionPage_example.py, openai_register.py, batch_open_nsfw.py """ -import tkinter as tk -from tkinter import ttk, messagebox, scrolledtext +try: + import tkinter as tk + from tkinter import ttk, messagebox, scrolledtext + TK_AVAILABLE = True + TK_IMPORT_ERROR = None +except ImportError as exc: + tk = None + ttk = None + messagebox = None + scrolledtext = None + TK_AVAILABLE = False + TK_IMPORT_ERROR = exc import threading import datetime import time @@ -58,11 +68,11 @@ DEFAULT_CONFIG = { "cloudmail_public_token": "", "cloudmail_domains": "", "cloudmail_path_messages": "/api/public/emailList", - "proxy": "http://127.0.0.1:7890", + "proxy": "", "enable_nsfw": True, "register_count": 1, "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", - "grok2api_auto_add_local": True, + "grok2api_auto_add_local": False, "grok2api_local_token_file": "", "grok2api_pool_name": "ssoBasic", "grok2api_auto_add_remote": False, @@ -100,9 +110,15 @@ def load_config(): 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} - except Exception: - config = DEFAULT_CONFIG.copy() + except Exception as exc: + message = f"配置文件解析失败: {CONFIG_FILE}: {exc}" + print(f"[!] {message}", file=sys.stderr) + raise SystemExit(message) + else: + config = DEFAULT_CONFIG.copy() return config @@ -647,38 +663,68 @@ def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None): token = _normalize_sso_token(raw_token) if not token: return False - token_file = resolve_grok2api_local_token_file() - pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() - if not pool_name: - pool_name = "ssoBasic" - os.makedirs(os.path.dirname(token_file), exist_ok=True) - data = {} - if os.path.exists(token_file): - try: - with open(token_file, "r", encoding="utf-8") as f: - data = json.load(f) or {} - except Exception: - data = {} - if not isinstance(data, dict): + token_file = os.path.abspath(resolve_grok2api_local_token_file()) + pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() or "ssoBasic" + parent = os.path.dirname(token_file) + os.makedirs(parent, exist_ok=True) + lock_path = token_file + ".lock" + try: + from filelock import FileLock + except Exception as exc: + raise RuntimeError(f"filelock 依赖不可用，拒绝非原子写入 token 池: {exc}") + with FileLock(lock_path, timeout=30): data = {} - pool = data.get(pool_name) - if not isinstance(pool, list): - pool = [] - 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", ""))) - if token in existing: - if log_callback: - log_callback(f"[*] grok2api 本地池已存在 token: {pool_name}") - return True - entry = {"token": token, "tags": ["auto-register"], "note": email} - pool.append(entry) - data[pool_name] = pool - with open(token_file, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=2) + if os.path.exists(token_file): + try: + with open(token_file, "r", encoding="utf-8") as f: + data = json.load(f) or {} + except Exception as exc: + broken_path = token_file + f".broken-{int(time.time())}" + try: + os.replace(token_file, broken_path) + except Exception: + broken_path = token_file + raise RuntimeError(f"本地 token 文件 JSON 解析失败，已停止写入以避免覆盖: {broken_path}: {exc}") + if not isinstance(data, dict): + raise RuntimeError("本地 token 文件根节点不是 JSON object，拒绝覆盖") + pool = data.get(pool_name) + if not isinstance(pool, list): + pool = [] + 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", ""))) + if token in existing: + if log_callback: + log_callback(f"[*] grok2api 本地池已存在 token: {pool_name}") + return True + pool.append({"token": token, "tags": ["auto-register"], "note": email}) + data[pool_name] = pool + if os.path.exists(token_file): + backup_path = token_file + ".bak" + try: + with open(token_file, "rb") as src, open(backup_path, "wb") as dst: + dst.write(src.read()) + dst.flush() + os.fsync(dst.fileno()) + except Exception as exc: + raise RuntimeError(f"创建本地 token 备份失败，拒绝继续写入: {exc}") + temp_path = token_file + ".tmp" + try: + with open(temp_path, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + f.write("\n") + f.flush() + os.fsync(f.fileno()) + os.replace(temp_path, token_file) + finally: + if os.path.exists(temp_path): + try: + os.unlink(temp_path) + except Exception: + pass if log_callback: log_callback(f"[+] 已写入 grok2api 本地池: {pool_name} ({token_file})") return True @@ -751,21 +797,30 @@ def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None): 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 + load_errors = [] for api_base in api_bases or [base]: try: resp = http_get(f"{api_base}/tokens", headers=headers, params=query, timeout=20) if resp.status_code == 200: payload = resp.json() - current = payload.get("tokens", {}) if isinstance(payload, dict) else {} - fallback_base = api_base - break - except Exception: - continue - if not isinstance(current, dict): - current = {} + 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}") + except Exception as exc: + load_errors.append(f"{api_base}/tokens: {exc}") + if not loaded_remote_state: + raise RuntimeError("无法安全读取远端 token 池，拒绝执行全量覆盖: " + "; ".join(load_errors)) pool = current.get(pool_name) if not isinstance(pool, list): pool = [] @@ -874,7 +929,7 @@ def http_post(url, **kwargs): def raise_if_cancelled(cancel_callback=None): if cancel_callback and cancel_callback(): - raise RegistrationCancelled("鐢ㄦ埛鍋滄娉ㄥ唽") + raise RegistrationCancelled("用户停止注册") def sleep_with_cancel(seconds, cancel_callback=None): @@ -1063,7 +1118,7 @@ def yyds_create_account(address=None, domain=None, api_key=None, jwt=None): data = resp.json() if data.get("success"): return data.get("data", {}) - raise Exception(f"YYDS 鍒涘缓閭澶辫触: {data}") + raise Exception(f"YYDS 创建邮箱失败: {data}") def yyds_get_token(address, api_key=None, jwt=None): @@ -1809,7 +1864,7 @@ def tk_entry(parent, textvariable=None, width=30, **kwargs): ) -def tk_button(parent, text="", command=None, state=tk.NORMAL, **kwargs): +def tk_button(parent, text="", command=None, state="normal", **kwargs): return tk.Button( parent, text=text, @@ -2822,6 +2877,8 @@ def wait_for_sso_cookie(timeout=120, log_callback=None, cancel_callback=None): final_no_submit_state = "" final_no_submit_since = None final_no_submit_timeout = 25 + last_wait_exception_message = "" + last_wait_exception_at = 0.0 while time.time() < deadline: raise_if_cancelled(cancel_callback) @@ -2948,8 +3005,14 @@ return String(cfInput.value || '').trim().length; refresh_active_page() except AccountRetryNeeded: raise - except Exception: - pass + except Exception as exc: + if log_callback: + now = time.time() + message = f"{exc.__class__.__name__}: {exc}" + if message != last_wait_exception_message or now - last_wait_exception_at >= 10: + log_callback(f"[Debug] 等待 sso cookie 时出现异常，将继续等待: {message}") + last_wait_exception_message = message + last_wait_exception_at = now sleep_with_cancel(1, cancel_callback) @@ -3648,6 +3711,10 @@ def main(): if len(sys.argv) > 1 and sys.argv[1].strip().lower() in ("start", "cli", "--cli"): main_cli() return + if not TK_AVAILABLE: + print(f"[!] GUI 模式需要 Tkinter，但当前环境不可用: {TK_IMPORT_ERROR}", file=sys.stderr) + print("[*] 可改用 CLI 模式: python grok_register_ttk.py cli", file=sys.stderr) + return root = tk.Tk() setup_light_theme(root) app = GrokRegisterGUI(root) diff --git a/requirements.txt b/requirements.txt index f0a06f5..b052add 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,7 +5,7 @@ cssselect==1.3.0 curl_cffi==0.13.0 DataRecorder==3.6.2 DownloadKit==2.0.7 -DrissionPage==4.1.1.2 +DrissionPage>=4.1.1.2,<4.2 filelock==3.19.1 idna==3.13 lxml==6.1.0 diff --git a/tools/apply_audit_fixes.py b/tools/apply_audit_fixes.py deleted file mode 100644 index fda3a1b..0000000 --- a/tools/apply_audit_fixes.py +++ /dev/null @@ -1,643 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -"""Apply targeted audit fixes without rewriting the project structure. - -This script is intentionally idempotent because the repository may already -contain some of the fixes from earlier interrupted bot runs. -""" - -from pathlib import Path -import ast -import re - -ROOT = Path(__file__).resolve().parents[1] -APP = ROOT / "grok_register_ttk.py" -REQ = ROOT / "requirements.txt" -GITIGNORE = ROOT / ".gitignore" -README = ROOT / "README.md" -OAUTH = ROOT / "cpa_xai" / "oauth_device.py" -BROWSER = ROOT / "cpa_xai" / "browser_confirm.py" -SCHEMA = ROOT / "cpa_xai" / "schema.py" -WRITER = ROOT / "cpa_xai" / "writer.py" - - -def read(path, encoding="utf-8"): - return path.read_text(encoding=encoding) - - -def write(path, content, encoding="utf-8"): - path.write_text(content, encoding=encoding) - - -def replace_required(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_if_present(text, old, new): - return text.replace(old, new, 1) if old in text else text - - -def replace_func_between(text, start_marker, end_marker, new_block, label): - start = text.find(start_marker) - if start < 0: - if new_block in text: - return text - raise RuntimeError(f"{label}: start marker not found") - end = text.find(end_marker, start) - if end < 0: - raise RuntimeError(f"{label}: end marker not found") - return text[:start] + new_block + text[end:] - - -def ensure_line(content, line): - lines = content.splitlines() - if line not in lines: - content = content.rstrip() + "\n" + line + "\n" - return content - - -# Trigger token: audit-fixes-v4 -# --------------------------------------------------------------------------- -# Main application fixes. -# --------------------------------------------------------------------------- -app = read(APP, encoding="utf-8-sig") - -# L-01 / M-05: UTF-8 without BOM and CLI can start without Tkinter installed. -app = replace_if_present( - app, - "import tkinter as tk\nfrom tkinter import ttk, messagebox, scrolledtext\n", - "try:\n import tkinter as tk\n from tkinter import ttk, messagebox, scrolledtext\n TK_AVAILABLE = True\n TK_IMPORT_ERROR = None\nexcept ImportError as exc:\n tk = None\n ttk = None\n messagebox = None\n scrolledtext = None\n TK_AVAILABLE = False\n TK_IMPORT_ERROR = exc\n", -) -app = app.replace( - 'def tk_button(parent, text="", command=None, state=tk.NORMAL, **kwargs):', - 'def tk_button(parent, text="", command=None, state="normal", **kwargs):', -) - -# M-02: safer defaults, matching config.example.json. -app = app.replace(' "proxy": "http://127.0.0.1:7890",', ' "proxy": "",') -app = app.replace(' "grok2api_auto_add_local": True,', ' "grok2api_auto_add_local": False,') -old_load_config = '''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 = {**DEFAULT_CONFIG, **loaded} - except Exception: - config = DEFAULT_CONFIG.copy() - return config -''' -new_load_config = '''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} - except Exception as exc: - message = f"配置文件解析失败: {CONFIG_FILE}: {exc}" - print(f"[!] {message}", file=sys.stderr) - raise SystemExit(message) - else: - config = DEFAULT_CONFIG.copy() - return config -''' -app = replace_if_present(app, old_load_config, new_load_config) - -# H-05 / M-01: local token writes are locked, backed up and atomic. -new_local_pool = '''def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None): - token = _normalize_sso_token(raw_token) - if not token: - return False - token_file = os.path.abspath(resolve_grok2api_local_token_file()) - pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() or "ssoBasic" - parent = os.path.dirname(token_file) - os.makedirs(parent, exist_ok=True) - lock_path = token_file + ".lock" - try: - from filelock import FileLock - except Exception as exc: - raise RuntimeError(f"filelock 依赖不可用，拒绝非原子写入 token 池: {exc}") - with FileLock(lock_path, timeout=30): - data = {} - if os.path.exists(token_file): - try: - with open(token_file, "r", encoding="utf-8") as f: - data = json.load(f) or {} - except Exception as exc: - broken_path = token_file + f".broken-{int(time.time())}" - try: - os.replace(token_file, broken_path) - except Exception: - broken_path = token_file - raise RuntimeError(f"本地 token 文件 JSON 解析失败，已停止写入以避免覆盖: {broken_path}: {exc}") - if not isinstance(data, dict): - raise RuntimeError("本地 token 文件根节点不是 JSON object，拒绝覆盖") - pool = data.get(pool_name) - if not isinstance(pool, list): - pool = [] - 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", ""))) - if token in existing: - if log_callback: - log_callback(f"[*] grok2api 本地池已存在 token: {pool_name}") - return True - pool.append({"token": token, "tags": ["auto-register"], "note": email}) - data[pool_name] = pool - if os.path.exists(token_file): - backup_path = token_file + ".bak" - try: - with open(token_file, "rb") as src, open(backup_path, "wb") as dst: - dst.write(src.read()) - dst.flush() - os.fsync(dst.fileno()) - except Exception as exc: - raise RuntimeError(f"创建本地 token 备份失败，拒绝继续写入: {exc}") - temp_path = token_file + ".tmp" - try: - with open(temp_path, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=2) - f.write("\n") - f.flush() - os.fsync(f.fileno()) - os.replace(temp_path, token_file) - finally: - if os.path.exists(temp_path): - try: - os.unlink(temp_path) - except Exception: - pass - if log_callback: - log_callback(f"[+] 已写入 grok2api 本地池: {pool_name} ({token_file})") - return True - - -''' -if "with FileLock(lock_path" not in app: - app = replace_func_between( - app, - "def add_token_to_grok2api_local_pool(raw_token, email=\"\", log_callback=None):\n", - "def get_grok2api_remote_api_bases(base):\n", - new_local_pool, - "atomic local token pool", - ) - -# H-01: remote fallback must not POST a full replacement unless old state was read. -old_remote_fallback = ''' # 兜底：旧版全量保存接口 - current = {} - fallback_base = api_bases[0] if api_bases else base - for api_base in api_bases or [base]: - try: - resp = http_get(f"{api_base}/tokens", headers=headers, params=query, timeout=20) - if resp.status_code == 200: - payload = resp.json() - current = payload.get("tokens", {}) if isinstance(payload, dict) else {} - fallback_base = api_base - break - except Exception: - continue - if not isinstance(current, dict): - current = {} -''' -new_remote_fallback = ''' # 兜底：旧版全量保存接口。必须先成功读取远端旧状态，避免空池覆盖。 - current = {} - fallback_base = api_bases[0] if api_bases else base - loaded_remote_state = False - load_errors = [] - for api_base in api_bases or [base]: - 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}") - except Exception as exc: - load_errors.append(f"{api_base}/tokens: {exc}") - if not loaded_remote_state: - raise RuntimeError("无法安全读取远端 token 池，拒绝执行全量覆盖: " + "; ".join(load_errors)) -''' -app = replace_if_present(app, old_remote_fallback, new_remote_fallback) - -# M-06: do not hide unexpected SSO wait exceptions forever. -if "last_wait_exception_message" not in app: - app = replace_if_present( - app, - ''' final_no_submit_state = "" - final_no_submit_since = None - final_no_submit_timeout = 25 - - while time.time() < deadline: -''', - ''' final_no_submit_state = "" - final_no_submit_since = None - final_no_submit_timeout = 25 - last_wait_exception_message = "" - last_wait_exception_at = 0.0 - - while time.time() < deadline: -''', - ) - app = replace_if_present( - app, - ''' except Exception: - pass - - sleep_with_cancel(1, cancel_callback) -''', - ''' except Exception as exc: - if log_callback: - now = time.time() - message = f"{exc.__class__.__name__}: {exc}" - if message != last_wait_exception_message or now - last_wait_exception_at >= 10: - log_callback(f"[Debug] 等待 sso cookie 时出现异常，将继续等待: {message}") - last_wait_exception_message = message - last_wait_exception_at = now - - sleep_with_cancel(1, cancel_callback) -''', - ) - -# L-02: repair known mojibake that affects logs/control flow. -for bad, good in { - "鐢ㄦ埛鍋滄娉ㄥ唽": "用户停止注册", - "YYDS 鍒涘缓閭澶辫触": "YYDS 创建邮箱失败", - "YYDS 鑾峰彇JWT澶辫触": "YYDS 获取 JWT 失败", -}.items(): - app = app.replace(bad, good) - -# M-05: GUI-only Tk failure should not break CLI mode. -old_main = '''def main(): - if len(sys.argv) > 1 and sys.argv[1].strip().lower() in ("start", "cli", "--cli"): - main_cli() - return - root = tk.Tk() - setup_light_theme(root) - app = GrokRegisterGUI(root) - root.mainloop() -''' -new_main = '''def main(): - if len(sys.argv) > 1 and sys.argv[1].strip().lower() in ("start", "cli", "--cli"): - main_cli() - return - if not TK_AVAILABLE: - print(f"[!] GUI 模式需要 Tkinter，但当前环境不可用: {TK_IMPORT_ERROR}", file=sys.stderr) - print("[*] 可改用 CLI 模式: python grok_register_ttk.py cli", file=sys.stderr) - return - root = tk.Tk() - setup_light_theme(root) - app = GrokRegisterGUI(root) - root.mainloop() -''' -app = replace_if_present(app, old_main, new_main) - -ast.parse(app) -write(APP, app, encoding="utf-8") - -# --------------------------------------------------------------------------- -# CPA browser / OAuth fixes. -# --------------------------------------------------------------------------- -browser = read(BROWSER) -browser = browser.replace("except BaseException as exc:", "except Exception as exc:") -old_proxy_block = ''' from .proxyutil import proxy_for_chromium, proxy_log_label, resolve_proxy - - resolved = resolve_proxy(proxy) - chrome_proxy = proxy_for_chromium(resolved) - 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) - page = browser.latest_tab - logger("standalone chromium started") - return browser, page -''' -new_proxy_block = ''' from .proxyutil import prepare_chromium_proxy, proxy_log_label, resolve_proxy - - resolved = resolve_proxy(proxy) - proxy_bridge = None - chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger) - 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 -''' -browser = replace_if_present(browser, old_proxy_block, new_proxy_block) -old_close_block = '''def close_standalone(browser: Any) -> None: - try: - browser.quit() - except Exception: - pass - - -_mint_tls = threading.local() -''' -new_close_block = '''def close_standalone(browser: Any) -> None: - if browser is None: - return - _unregister_mint_browser(browser) - bridge = getattr(browser, "_cpa_proxy_bridge", None) - try: - browser.quit() - except Exception: - pass - if bridge is not None: - try: - bridge.stop() - except Exception: - pass - - -_mint_tls = threading.local() -_mint_registry_lock = threading.Lock() -_mint_registry = set() - - -def _register_mint_browser(browser: Any) -> None: - if browser is None: - return - with _mint_registry_lock: - _mint_registry.add(browser) - - -def _unregister_mint_browser(browser: Any) -> None: - if browser is None: - return - with _mint_registry_lock: - _mint_registry.discard(browser) -''' -if "_mint_registry = set()" not in browser: - browser = replace_required(browser, old_close_block, new_close_block, "CPA global browser registry") -old_shutdown = '''def shutdown_mint_browsers() -> None: - state = _mint_tls_get() - browser = state.get("browser") - if browser is not None: - try: - close_standalone(browser) - except Exception: - pass - state.update({"browser": None, "page": None, "served": 0, "proxy": None, "headless": None}) -''' -new_shutdown = '''def shutdown_mint_browsers() -> None: - state = _mint_tls_get() - with _mint_registry_lock: - browsers = list(_mint_registry) - for browser in browsers: - try: - close_standalone(browser) - except Exception: - pass - state.update({"browser": None, "page": None, "served": 0, "proxy": None, "headless": None}) -''' -browser = replace_if_present(browser, old_shutdown, new_shutdown) -old_poll = ''' def _poll() -> None: - try: - time.sleep(2) - result = poll_device_token( - session.device_code, - token_endpoint=session.token_endpoint, - interval=max(session.interval, 5), - expires_in=min(session.expires_in, int(browser_timeout_sec) + 60), - log=logger, - cancel=cancel, - proxy=resolved or None, - ) - token_box["token"] = result - stop_event.set() - logger("token poll SUCCESS — stop_event set") - except Exception as exc: - error_box["err"] = exc - stop_event.set() -''' -new_poll = ''' def combined_cancel(): - return stop_event.is_set() or bool(cancel and cancel()) - - def _poll() -> None: - try: - for _ in range(20): - if combined_cancel(): - raise OAuthDeviceError("cancelled") - time.sleep(0.1) - result = poll_device_token( - session.device_code, - token_endpoint=session.token_endpoint, - interval=max(session.interval, 5), - expires_in=min(session.expires_in, int(browser_timeout_sec) + 60), - log=logger, - cancel=combined_cancel, - proxy=resolved or None, - ) - token_box["token"] = result - stop_event.set() - logger("token poll SUCCESS — stop_event set") - except Exception as exc: - error_box["err"] = exc - stop_event.set() -''' -browser = replace_if_present(browser, old_poll, new_poll) -old_join = ''' if hard: - stop_event.set() - raise - thread.join(timeout=max(browser_timeout_sec, 60) + 30) -''' -new_join = ''' if hard: - stop_event.set() - thread.join(timeout=5) - if thread.is_alive(): - logger("token poll thread did not stop within 5s after browser failure") - raise - thread.join(timeout=max(browser_timeout_sec, 60) + 30) - if thread.is_alive(): - stop_event.set() - thread.join(timeout=5) - if thread.is_alive(): - raise OAuthDeviceError("token poll thread did not stop after timeout") -''' -browser = replace_if_present(browser, old_join, new_join) -ast.parse(browser) -write(BROWSER, browser) - -oauth = read(OAUTH) -oauth = oauth.replace("except BaseException as exc:", "except Exception as exc:") -old_poll_header = '''def poll_device_token( - device_code, - token_endpoint, - client_id=CLIENT_ID, - interval=5, - expires_in=1800, - timeout=30.0, - log=None, - cancel=None, - proxy=None, -): - logger = log or (lambda message: None) - deadline = time.time() + max(int(expires_in) - 5, 30) - sleep_seconds = max(int(interval), 1) - net_streak = 0 - max_net_streak = 20 - while time.time() < deadline: - if cancel and cancel(): - raise OAuthDeviceError("cancelled") -''' -new_poll_header = '''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, - client_id=CLIENT_ID, - interval=5, - expires_in=1800, - timeout=30.0, - log=None, - cancel=None, - proxy=None, -): - logger = log or (lambda message: None) - deadline = time.time() + max(int(expires_in) - 5, 30) - sleep_seconds = max(int(interval), 1) - net_streak = 0 - max_net_streak = 20 - while time.time() < deadline: - if cancel and cancel(): - raise OAuthDeviceError("cancelled") -''' -if "def _sleep_with_cancel(" not in oauth: - oauth = replace_required(oauth, old_poll_header, new_poll_header, "oauth cancellable sleep helper") - oauth = oauth.replace("time.sleep(wait_seconds)", "_sleep_with_cancel(wait_seconds, cancel)") - oauth = oauth.replace("time.sleep(sleep_seconds)", "_sleep_with_cancel(sleep_seconds, cancel)") -ast.parse(oauth) -write(OAUTH, oauth) - -# --------------------------------------------------------------------------- -# CPA schema/writer hardening. -# --------------------------------------------------------------------------- -schema = read(SCHEMA) -old_expired = ''' expired = "" - if exp: - expired = datetime.fromtimestamp(exp, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") -''' -new_expired = ''' expired = "" - if exp: - expired = datetime.fromtimestamp(exp, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - elif expires_in: - expired = datetime.fromtimestamp(time.time() + int(expires_in or 0), tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") -''' -schema = replace_if_present(schema, old_expired, new_expired) -ast.parse(schema) -write(SCHEMA, schema) - -writer = read(WRITER) -old_writer_head = '''def write_cpa_xai_auth(auth_dir, payload, filename=None): - root = Path(auth_dir).expanduser().resolve() - root.mkdir(parents=True, exist_ok=True) - target_name = filename or credential_file_name(payload.get("email", ""), payload.get("sub", "")) - if not str(target_name).endswith(".json"): - target_name = str(target_name) + ".json" - destination = root / str(target_name) -''' -new_writer_head = '''def _is_relative_to(path, root): - try: - path.relative_to(root) - return True - except ValueError: - return False - - -def write_cpa_xai_auth(auth_dir, payload, filename=None): - root = Path(auth_dir).expanduser().resolve() - root.mkdir(parents=True, exist_ok=True) - target_name = filename or credential_file_name(payload.get("email", ""), payload.get("sub", "")) - target_name = Path(str(target_name)).name - if not str(target_name).endswith(".json"): - target_name = str(target_name) + ".json" - destination = (root / str(target_name)).resolve() - if not _is_relative_to(destination, root): - raise ValueError("CPA auth filename must stay inside auth_dir") -''' -writer = replace_if_present(writer, old_writer_head, new_writer_head) -ast.parse(writer) -write(WRITER, writer) - -# --------------------------------------------------------------------------- -# Requirements / ignore / docs. -# --------------------------------------------------------------------------- -req = read(REQ) -req = req.replace("DrissionPage==4.1.1.2", "DrissionPage>=4.1.1.2,<4.2") -write(REQ, req) - -gitignore = read(GITIGNORE) -gitignore = ensure_line(gitignore, "screenshots/") -gitignore = ensure_line(gitignore, "*.png") -write(GITIGNORE, gitignore) - -readme = read(README) -readme = readme.replace( - "- `cpa_auths/cpa_auth_failed.txt`：OIDC 导出失败记录。\n- `*.log`：可选日志文件。", - "- `cpa_auths/cpa_auth_failed.txt`：OIDC 导出失败记录。\n- `screenshots/`：CPA/OIDC 浏览器失败调试截图，已被 `.gitignore` 忽略。\n- `*.log`：可选日志文件。", -) -old_tree = '''```text -. -├── grok_register_ttk.py # 主程序 -├── cf_mail_debug.py # Cloudflare 邮箱调试工具 -├── config.example.json # 配置示例 -├── requirements.txt # Python 依赖 -└── README.md -``` -''' -new_tree = '''```text -. -├── grok_register_ttk.py # 主程序（GUI / CLI） -├── cpa_export.py # 注册成功后的 CPA/OIDC 导出入口 -├── cpa_xai/ # xAI Device Auth、浏览器授权和凭证写入模块 -├── cf_mail_debug.py # Cloudflare 邮箱调试工具 -├── config.example.json # 配置示例 -├── requirements.txt # Python 依赖 -├── tests/ # 现有测试用例 -├── assets/ # README 资源 -└── README.md -``` -''' -readme = replace_if_present(readme, old_tree, new_tree) -write(README, readme) - -# Final syntax validation for touched Python files. -for path in (APP, OAUTH, BROWSER, SCHEMA, WRITER): - ast.parse(read(path, encoding="utf-8")) - -print("audit fixes applied") diff --git a/tools/fix_audit_patch_escapes.py b/tools/fix_audit_patch_escapes.py deleted file mode 100644 index 8a9e536..0000000 --- a/tools/fix_audit_patch_escapes.py +++ /dev/null @@ -1,22 +0,0 @@ -#!/usr/bin/env python3 -"""Repair string escapes inside the temporary audit patch script before running it. - -The patch script embeds Python source inside triple-quoted strings. Source snippets -that should generate literal backslash-n in grok_register_ttk.py must therefore -use double escaping in this script. -""" -from pathlib import Path - -path = Path(__file__).resolve().parent / "apply_audit_fixes.py" -text = path.read_text(encoding="utf-8") - -replacements = { - 'f.write("\\n")': 'f.write("\\\\n")', - 'line = f"{email}----{password or \'\'}----{sso}\\n"': 'line = f"{email}----{password or \'\'}----{sso}\\\\n"', -} - -for old, new in replacements.items(): - text = text.replace(old, new) - -path.write_text(text, encoding="utf-8") -print("audit patch escapes repaired")