776 lines
27 KiB
Python
776 lines
27 KiB
Python
#!/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")
|