codex-oauth-plus-onboarding-skill/scripts/codex_oauth_onboarding_runner.py

#!/usr/bin/env python3
"""
Token-lite Codex OAuth onboarding runner skeleton.

Purpose: keep future Codex2API/OpenAI/5sim onboarding runs script-driven with compact JSONL events.
This script intentionally centralizes config parsing, redaction, state classification, and event logging.
Sensitive values are never printed directly.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import pathlib
import re
import shlex
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from typing import Any, Dict, Iterable, List, Optional, Tuple

COUNTRY_META = {
    "argentina": {"label_zh": "阿根廷", "prefix": "54"},
    "netherlands": {"label_zh": "荷兰", "prefix": "31"},
    "indonesia": {"label_zh": "印度尼西亚", "prefix": "62"},
    "vietnam": {"label_zh": "越南", "prefix": "84"},
    "england": {"label_zh": "英国", "prefix": "44"},
    "japan": {"label_zh": "日本", "prefix": "81"},
    "germany": {"label_zh": "德国", "prefix": "49"},
}

SECRET_KEY_RE = re.compile(r"(KEY|TOKEN|SECRET|PASSWORD|PASS|CODE|CALLBACK)", re.I)
DIRECT_PHONE_REJECT = "无法向此电话号码发送验证码"
INVALID_PHONE = "电话号码无效"
ACCOUNT_DEACTIVATED = "account_deactivated"


def now() -> str:
    return dt.datetime.now().isoformat(timespec="seconds")


def load_env(path: pathlib.Path) -> Dict[str, str]:
    env: Dict[str, str] = {}
    for raw in path.read_text().splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        env[k.strip()] = v.strip().strip('"\'')
    return env


def redact_key_value(key: str, value: Any) -> Any:
    if value is None:
        return None
    if SECRET_KEY_RE.search(key):
        return "[REDACTED]" if str(value) else ""
    if isinstance(value, str) and "code=" in value and "state=" in value:
        return re.sub(r"code=[^&]+", "code=[REDACTED]", value)
    return value


def redact_event(event: Dict[str, Any]) -> Dict[str, Any]:
    return {k: redact_key_value(k, v) for k, v in event.items()}


class RunLog:
    def __init__(self, run_dir: pathlib.Path):
        self.run_dir = run_dir
        self.run_dir.mkdir(parents=True, exist_ok=True)
        (self.run_dir / "screenshots").mkdir(exist_ok=True)
        self.events_path = self.run_dir / "events.jsonl"

    def emit(self, phase: str, result: str = "info", **fields: Any) -> None:
        event = {"ts": now(), "phase": phase, "result": result, **fields}
        safe = redact_event(event)
        with self.events_path.open("a") as f:
            f.write(json.dumps(safe, ensure_ascii=False, separators=(",", ":")) + "\n")
        print(json.dumps(safe, ensure_ascii=False, separators=(",", ":")), flush=True)


def run(cmd: str, timeout: int = 30, input_text: Optional[str] = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["bash", "-lc", cmd],
        input=input_text,
        text=True,
        capture_output=True,
        timeout=timeout,
    )


def cdp_eval(js: str, match: str = "auth.openai.com", timeout: int = 45) -> str:
    proc = run(f"CDP_MATCH={shlex.quote(match)} python3 /tmp/cdp_async_eval.py", timeout=timeout, input_text=js)
    return (proc.stdout or "") + (proc.stderr or "")


def classify_page(raw: str) -> str:
    if ACCOUNT_DEACTIVATED in raw:
        return "account_deactivated"
    if "Cloudflare" in raw or "验证您是真人" in raw or "challenge" in raw.lower():
        return "cloudflare_or_challenge"
    if "auth.openai.com/log-in" in raw or "欢迎回来" in raw:
        return "login_page"
    if "email-verification" in raw or "检查你的收件箱" in raw:
        return "email_code_page"
    if "add-phone" in raw or "电话号码是必填项" in raw:
        return "add_phone_page"
    if "重新发送" in raw and ("验证码" in raw or "代码" in raw):
        return "phone_code_page"
    if DIRECT_PHONE_REJECT in raw:
        return "phone_direct_reject"
    return "unknown"


def country_order(env: Dict[str, str]) -> List[str]:
    order = [x.strip().lower() for x in env.get("FIVE_SIM_COUNTRY_ORDER", "").split(",") if x.strip()]
    if order:
        return order
    country_id = env.get("FIVE_SIM_COUNTRY_ID", "").strip().lower()
    if country_id:
        return [country_id]
    return ["vietnam"]


def codex2api_origin(env: Dict[str, str]) -> str:
    raw = env.get("CODEX2API_URL", "").strip().rstrip("/")
    u = urllib.parse.urlsplit(raw)
    return f"{u.scheme}://{u.netloc}" if u.scheme and u.netloc else raw


def generate_codex2api_oauth_url(env: Dict[str, str]) -> Dict[str, Any]:
    origin = codex2api_origin(env)
    req = urllib.request.Request(
        origin + "/api/admin/oauth/generate-auth-url",
        data=b"{}",
        headers={"X-Admin-Key": env.get("CODEX2API_ADMIN_KEY", ""), "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=20) as resp:
        return json.loads(resp.read().decode())


def open_cdp_url(url: str) -> None:
    req = urllib.request.Request("http://127.0.0.1:9223/json/new?" + urllib.parse.quote(url, safe=""), method="PUT")
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()


def probe_proxy(proxy: str) -> Tuple[bool, str]:
    if proxy.startswith("socks5://"):
        curl_proxy = "socks5h://" + proxy[len("socks5://") :]
    else:
        curl_proxy = proxy
    proc = run(f"curl --proxy {shlex.quote(curl_proxy)} --max-time 8 https://api.ipify.org?format=json", timeout=12)
    ok = proc.returncode == 0 and "ip" in proc.stdout
    return ok, proc.stdout.strip()[-120:]


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--env", default="~/.hermes/env/codex_oauth_onboarding.env")
    ap.add_argument("--run-root", default="~/.Hermes/workspace/runs/codex-oauth")
    ap.add_argument("--print-plan", action="store_true", help="Only print resolved low-token plan/config summary.")
    args = ap.parse_args()

    env_path = pathlib.Path(args.env).expanduser()
    env = load_env(env_path)
    run_id = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
    log = RunLog(pathlib.Path(args.run_root).expanduser() / run_id)

    countries = country_order(env)
    proxy = env.get("BROWSER_PROXY_SERVER", "")
    replacement_limit = int(env.get("PHONE_VERIFICATION_REPLACEMENT_LIMIT") or "3")
    log.emit(
        "preflight",
        "config_loaded",
        env_path=str(env_path),
        target_mode=env.get("PANEL_MODE") or env.get("OAUTH_TARGET_MODE") or "codex2api",
        proxy=proxy,
        country_order=",".join(countries),
        replacement_limit=replacement_limit,
        five_sim_product=env.get("FIVE_SIM_PRODUCT") or env.get("FIVE_SIM_SERVICE") or "openai",
        operator=env.get("FIVE_SIM_OPERATOR") or "any",
    )

    if proxy:
        ok, info = probe_proxy(proxy)
        log.emit("preflight", "proxy_ok" if ok else "proxy_failed", proxy=proxy, info=info)
        if not ok:
            return 2

    if args.print_plan:
        log.emit("plan", "ready", next="run full automation script implementation/resume")
        return 0

    log.emit(
        "not_implemented",
        "skeleton_only",
        message="This script currently provides the token-lite logging/config skeleton. Add browser/email/phone step execution before using as the primary runner.",
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())