From 8f96b9838d71f90b907d0d60861bf30768d3da9f Mon Sep 17 00:00:00 2001 From: Aaron Liang <76561968+AaronL725@users.noreply.github.com> Date: Tue, 14 Jul 2026 21:25:11 +0800 Subject: [PATCH] chore: add temporary audit fix patch script --- tools/apply_audit_fixes.py | 775 +++++++++++++++++++++++++++++++++++++ 1 file changed, 775 insertions(+) create mode 100644 tools/apply_audit_fixes.py diff --git a/tools/apply_audit_fixes.py b/tools/apply_audit_fixes.py new file mode 100644 index 0000000..10b017e --- /dev/null +++ b/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")