chore: add temporary audit fix patch script

tools/apply_audit_fixes.py

@@ -0,0 +1,775 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Apply targeted audit fixes without rewriting the project structure."""
+
+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_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_optional(text, old, new):
+    return text.replace(old, new, 1) if old in text else text
+
+
+# ---------------------------------------------------------------------------
+# Main application fixes.
+# ---------------------------------------------------------------------------
+app = read(APP, encoding="utf-8-sig")
+
+# L-01: rewrite later as UTF-8 without BOM. Also make CLI independent from Tk.
+app = replace_once(
+    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",
+    "tkinter lazy import",
+)
+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: fail closed on broken config and align safe defaults with config.example.
+app = app.replace('    "proxy": "http://127.0.0.1:7890",', '    "proxy": "",')
+app = app.replace('    "grok2api_auto_add_local": True,', '    "grok2api_auto_add_local": False,')
+app = replace_once(
+    app,
+    '''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
+''',
+    '''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
+''',
+    "load_config fail closed",
+)
+
+# H-05 / M-01: local token writes are now locked, backed up and atomic.
+app = replace_once(
+    app,
+    '''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):
+        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 log_callback:
+        log_callback(f"[+] 已写入 grok2api 本地池: {pool_name} ({token_file})")
+    return True
+''',
+    '''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"
+        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)
+    if log_callback:
+        log_callback(f"[+] 已写入 grok2api 本地池: {pool_name} ({token_file})")
+    return True
+''',
+    "atomic local token pool",
+)
+
+# H-01: remote fallback must not POST a full replacement unless old state was read.
+app = replace_once(
+    app,
+    '''    # 兜底：旧版全量保存接口
+    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 = {}
+''',
+    '''    # 兜底：旧版全量保存接口。必须先成功读取远端旧状态，避免空池覆盖。
+    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))
+''',
+    "remote fail closed fallback",
+)
+
+# M-06: do not hide unexpected SSO wait exceptions forever.
+app = replace_once(
+    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:
+''',
+    "wait_for_sso diagnostics vars",
+)
+app = replace_once(
+    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)
+''',
+    "wait_for_sso limited exception log",
+)
+
+# 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)
+
+# L-06: centralize the duplicated successful-account output side effects.
+if "def handle_successful_account_outputs(" not in app:
+    app = replace_once(
+        app,
+        '''    return result
+
+
+class GrokRegisterGUI:
+''',
+        '''    return result
+
+
+def handle_successful_account_outputs(email, password, sso, accounts_output_file, log_callback=None, cancel_callback=None):
+    logger = log_callback or (lambda message: None)
+    try:
+        line = f"{email}----{password or ''}----{sso}\n"
+        with open(accounts_output_file, "a", encoding="utf-8") as f:
+            f.write(line)
+    except Exception as file_exc:
+        logger(f"[Debug] 保存账号文件失败: {file_exc}")
+    add_token_to_grok2api_pools(sso, email=email, log_callback=logger)
+    maybe_export_cpa_xai_after_success(
+        email=email,
+        password=password or "",
+        sso=sso,
+        log_callback=logger,
+        cancel_callback=cancel_callback,
+    )
+
+
+class GrokRegisterGUI:
+''',
+        "success output helper",
+    )
+
+app = replace_once(
+    app,
+    '''                    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,
+                    )
+''',
+    '''                    handle_successful_account_outputs(
+                        email=email,
+                        password=profile.get("password", ""),
+                        sso=sso,
+                        accounts_output_file=self.accounts_output_file,
+                        log_callback=self.log,
+                        cancel_callback=self.should_stop,
+                    )
+''',
+    "GUI success output helper call",
+)
+app = replace_once(
+    app,
+    '''                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,
+                )
+''',
+    '''                handle_successful_account_outputs(
+                    email=email,
+                    password=profile.get("password", ""),
+                    sso=sso,
+                    accounts_output_file=accounts_output_file,
+                    log_callback=cli_log,
+                    cancel_callback=controller.should_stop,
+                )
+''',
+    "CLI success output helper call",
+)
+
+# M-05: GUI-only Tk failure should not break CLI mode.
+app = replace_once(
+    app,
+    '''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()
+''',
+    '''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()
+''',
+    "main Tk guard",
+)
+
+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:")
+browser = replace_once(
+    browser,
+    '''    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
+''',
+    '''    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
+''',
+    "CPA authenticated browser proxy bridge",
+)
+browser = replace_once(
+    browser,
+    '''def close_standalone(browser: Any) -> None:
+    try:
+        browser.quit()
+    except Exception:
+        pass
+
+
+_mint_tls = threading.local()
+''',
+    '''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)
+''',
+    "CPA global browser registry",
+)
+browser = replace_once(
+    browser,
+    '''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})
+''',
+    '''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})
+''',
+    "CPA shutdown all registered browsers",
+)
+browser = replace_once(
+    browser,
+    '''        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()
+''',
+    '''        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()
+''',
+    "CPA combined cancel",
+)
+browser = replace_once(
+    browser,
+    '''            if hard:
+                stop_event.set()
+                raise
+        thread.join(timeout=max(browser_timeout_sec, 60) + 30)
+''',
+    '''            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")
+''',
+    "CPA poll thread join on browser failure",
+)
+ast.parse(browser)
+write(BROWSER, browser)
+
+oauth = read(OAUTH)
+oauth = oauth.replace("except BaseException as exc:", "except Exception as exc:")
+oauth = replace_once(
+    oauth,
+    '''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")
+''',
+    '''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")
+''',
+    "oauth cancellable sleep helper",
+)
+for old in (
+    "time.sleep(wait_seconds)",
+    "time.sleep(sleep_seconds)",
+):
+    oauth = oauth.replace(old, old.replace("time.sleep", "_sleep_with_cancel").replace(")", ", cancel)"))
+ast.parse(oauth)
+write(OAUTH, oauth)
+
+# ---------------------------------------------------------------------------
+# CPA schema/writer hardening.
+# ---------------------------------------------------------------------------
+schema = read(SCHEMA)
+schema = replace_once(
+    schema,
+    '''    expired = ""
+    if exp:
+        expired = datetime.fromtimestamp(exp, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+''',
+    '''    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")
+''',
+    "CPA expired fallback",
+)
+ast.parse(schema)
+write(SCHEMA, schema)
+
+writer = read(WRITER)
+writer = replace_once(
+    writer,
+    '''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)
+''',
+    '''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")
+''',
+    "CPA writer path guard",
+)
+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)
+for line in ("screenshots/", "*.png"):
+    if line not in gitignore.splitlines():
+        gitignore = gitignore.rstrip() + "\n" + line + "\n"
+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`：可选日志文件。",
+)
+readme = replace_once(
+    readme,
+    '''```text
+.
+├── grok_register_ttk.py   # 主程序
+├── cf_mail_debug.py       # Cloudflare 邮箱调试工具
+├── config.example.json    # 配置示例
+├── requirements.txt       # Python 依赖
+└── README.md
+```
+''',
+    '''```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 directory 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")