#!/usr/bin/env python3
"""web_login.py - Trae CN Relay 本地授权回调监听器

使用方式:
  python web_login.py [--relay https://cn.jj2.top] [--port 8765]

--relay 的取值优先级（一般不需要手动填写）:
  1. 命令行 --relay
  2. 环境变量 TRAE_RELAY_URL
  3. 从 relay 网页端「下载授权助手」时自动写入的部署地址
  4. http://127.0.0.1:8000（本机默认部署）

依赖: 仅用 Python 标准库。
"""

import argparse
import http.server
import ipaddress
import json
import os
import re
import sys
import urllib.parse
import urllib.request
import uuid
import webbrowser
from pathlib import Path

# 本副本由中转站 https://cn.jj2.top 于 2026-10-27 23:31:14 自动生成。
# 中转站:   https://cn.jj2.top
# 监听端口: 8765
# 部署地址与监听端口(:8765)已写入下方常量，直接运行即可，无需修改 IP/端口。
# 如需指向其它中转站，使用: python web_login.py --relay http://<host>:<port>
# ---- 部署默认值（由中转站下载接口按当前部署自动生成，手动运行也可用 --relay 覆盖）
DEFAULT_RELAY_URL = "https://cn.jj2.top"
DEFAULT_LISTENER_PORT = 8765
DEFAULT_CLIENT_ID = "ono9krqynydwx5"
DEFAULT_AUTH_URL = "https://www.trae.cn/authorization"


def normalize_relay_url(value: str) -> str:
    """Normalize a relay base URL, including an optional reverse-proxy prefix."""
    raw = str(value or "").strip()
    if not raw:
        return ""
    if "://" not in raw:
        raw = "http://" + raw
    if len(raw) > 200 or any(
        ord(char) < 32 or ord(char) > 126 or char in "!%\"'`$<>|&;(){}*?# ,\\"
        for char in raw
    ):
        return ""
    try:
        parsed = urllib.parse.urlsplit(raw)
        port = parsed.port
    except ValueError:
        return ""
    if parsed.scheme not in ("http", "https") or not parsed.hostname:
        return ""
    if "@" in parsed.netloc or parsed.netloc.endswith(":") or parsed.query or parsed.fragment:
        return ""
    if port is not None and not 1 <= port <= 65535:
        return ""
    host = parsed.hostname.lower()
    if ":" in host:
        try:
            host = f"[{ipaddress.IPv6Address(host)}]"
        except ValueError:
            return ""
    elif not re.fullmatch(r"[a-z0-9][a-z0-9._-]*", host):
        return ""
    path = parsed.path.rstrip("/")
    if path and (
        not re.fullmatch(r"/[A-Za-z0-9._~/-]*", path)
        or any(part in (".", "..") for part in path.split("/"))
    ):
        return ""
    netloc = f"{host}:{port}" if port is not None else host
    return f"{parsed.scheme}://{netloc}{path}"


def default_relay_url() -> str:
    for candidate in (
        os.environ.get("TRAE_RELAY_URL", ""),
        DEFAULT_RELAY_URL,
    ):
        normalized = normalize_relay_url(candidate)
        if normalized:
            return normalized
    return normalize_relay_url(DEFAULT_RELAY_URL) or "http://127.0.0.1:8000"


def parse_args():
    parser = argparse.ArgumentParser(description="Trae CN Relay 本地授权回调监听器")
    parser.add_argument("--relay", default=default_relay_url(), help="Relay 地址")
    parser.add_argument("--port", type=int, default=DEFAULT_LISTENER_PORT, help="本地监听端口")
    parser.add_argument("--client-id", default=DEFAULT_CLIENT_ID, help="Trae Client ID")
    parser.add_argument("--auth-url", default=DEFAULT_AUTH_URL, help="Trae 授权 URL")
    parser.add_argument("--no-open", action="store_true", help="不自动打开浏览器")
    return parser.parse_args()


def json_loads_safe(value: str) -> dict:
    if not value:
        return {}
    try:
        return json.loads(value)
    except (json.JSONDecodeError, TypeError):
        return {}


def parse_oauth_params(query: dict) -> dict:
    """解析 Trae 授权回调参数。

    Trae 网页授权页实际上会走两套流程：
      1. 新流程 (code_challenge): callback 会带 authCodeInfo / code 等参数
      2. 老流程 (refreshToken): callback 直接带 refreshToken=xxx
    对于老流程，本地不会拥有 Cloud-IDE-JWT，需要拿到
    refreshToken 后通过 oauth/ExchangeToken 向 api.trae.cn 兑换 Cloud-IDE-JWT。
    """
    user_jwt = json_loads_safe(query.get("userJwt", ""))
    token = user_jwt.get("Token") or user_jwt.get("token") or ""
    refresh = user_jwt.get("RefreshToken") or user_jwt.get("refreshToken") or query.get("refreshToken") or query.get("data") or ""
    token_exp = user_jwt.get("TokenExpireAt") or user_jwt.get("tokenExpireAt") or ""
    refresh_exp = user_jwt.get("RefreshExpireAt") or user_jwt.get("refreshExpireAt") or query.get("refreshExpireAt") or ""
    user_info = json_loads_safe(query.get("userInfo", ""))
    user_id = user_info.get("UserID") or user_info.get("userId") or user_info.get("userID") or query.get("userId") or ""
    region = user_info.get("Region") or user_info.get("region") or query.get("region") or "CN"
    ai_region = user_info.get("AIRegion") or user_info.get("aiRegion") or region
    client_id = user_jwt.get("ClientID") or user_jwt.get("clientId") or query.get("clientID") or query.get("clientId") or query.get("client_id") or ""
    uid = user_id or ""
    host = query.get("host") or user_info.get("Host") or user_info.get("host") or ""

    # 老流程：回调只带了 refreshToken，通过 oauth/ExchangeToken 兑换
    # Cloud-IDE-JWT 和用户信息。
    if not token and refresh:
        exchange = exchange_refresh_token(
            refresh_token=refresh,
            client_id=client_id or getattr(RelayAuthHandler, "client_id", DEFAULT_CLIENT_ID),
            host=host,
        )
        if exchange.get("token"):
            token = exchange["token"]
            refresh = exchange.get("refreshToken") or refresh
            token_exp = exchange.get("expiredAt") or token_exp
            refresh_exp = exchange.get("refreshExpiredAt") or refresh_exp
            client_id = exchange.get("clientId") or client_id
            host = exchange.get("host") or host
            user_id = exchange.get("userId") or user_id
            user_info = exchange.get("userInfo") or user_info
            region = exchange.get("region") or region
            ai_region = exchange.get("aiRegion") or region

    if not token:
        return {}

    return {
        "token": token,
        "refreshToken": refresh,
        "userId": user_id or "",
        "tenantId": user_info.get("TenantID") or user_info.get("tenantId") or "",
        "region": region,
        "aiRegion": ai_region,
        "host": host,
        "expiredAt": str(token_exp) if token_exp else "",
        "refreshExpiredAt": str(refresh_exp) if refresh_exp else "",
        "clientId": client_id,
        "webId": user_info.get("WebId") or user_info.get("webId") or uid,
        "bizUserId": user_info.get("BizUserId") or user_info.get("bizUserId") or uid,
        "userUniqueId": user_info.get("UserUniqueId") or user_info.get("userUniqueId") or uid,
        "scope": query.get("scope") or user_info.get("Scope") or user_info.get("scope") or "",
        "tenant": user_info.get("Tenant") or user_info.get("tenant") or "",
        "appLanguage": user_info.get("AppLanguage") or user_info.get("appLanguage") or "",
        "userRegion": query.get("userRegion") or user_info.get("UserRegion") or user_info.get("userRegion") or "",
        "userIdentity": user_info.get("UserIdentity") or user_info.get("userIdentity") or "",
        "screenName": user_info.get("ScreenName") or user_info.get("screenName") or "",
    }

def exchange_refresh_token(refresh_token: str, client_id: str, host: str = "") -> dict:
    """使用 refreshToken 向 Trae CN 兑换 Cloud-IDE-JWT。

    这与 Trae 官网的实现一致:
      POST https://api.trae.cn/cloudide/api/v3/trae/oauth/ExchangeToken
      {"ClientID":..., "RefreshToken":..., "ClientSecret":"-", "UserID":""}
    """
    if not refresh_token:
        return {}
    base = host or "https://api.trae.cn"
    base = base.rstrip("/")
    url = base + "/cloudide/api/v3/trae/oauth/ExchangeToken"
    payload = {
        "ClientID": client_id,
        "RefreshToken": refresh_token,
        "ClientSecret": "-",
        "UserID": "",
    }
    try:
        req = urllib.request.Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
        )
        req.add_header("Content-Type", "application/json")
        req.add_header("Accept", "application/json")
        with urllib.request.urlopen(req, timeout=20) as resp:
            body = resp.read().decode("utf-8")
    except urllib.error.HTTPError as e:
        try:
            body = e.read().decode("utf-8", errors="replace")
        except Exception:
            body = ""
        print(f"[web_login] ExchangeToken HTTP {e.code}: {body[:500]}")
        return {}
    except Exception as e:
        print(f"[web_login] ExchangeToken failed: {e}")
        return {}

    result = json_loads_safe(body)
    data = result.get("Result") or result.get("result") or result
    if not isinstance(data, dict):
        print(f"[web_login] ExchangeToken unexpected result: {body[:500]}")
        return {}

    token = data.get("Token") or data.get("token") or data.get("AccessToken") or data.get("accessToken") or ""
    if not token:
        print(f"[web_login] ExchangeToken missing Token: {body[:500]}")
        return {}

    refresh2 = data.get("RefreshToken") or data.get("refreshToken") or refresh_token
    user_info = json_loads_safe(str(data.get("UserInfo") or data.get("userInfo") or ""))
    return {
        "token": token,
        "refreshToken": refresh2,
        "expiredAt": str(data.get("TokenExpireAt") or data.get("tokenExpireAt") or ""),
        "refreshExpiredAt": str(data.get("RefreshExpireAt") or data.get("refreshExpireAt") or ""),
        "clientId": data.get("ClientID") or data.get("clientId") or client_id,
        "host": host or base,
        "userId": user_info.get("UserID") or user_info.get("userId") or data.get("UserID") or data.get("userId") or "",
        "userInfo": user_info,
        "region": user_info.get("Region") or user_info.get("region") or data.get("Region") or data.get("region") or "CN",
        "aiRegion": user_info.get("AIRegion") or user_info.get("aiRegion") or data.get("AIRegion") or data.get("aiRegion") or "",
    }


def forward_to_relay(relay_url: str, creds: dict) -> tuple[bool, str]:
    """POST 凭证到 relay /api/web-auth，返回 (success, message)"""
    url = relay_url.rstrip("/") + "/api/web-auth"
    data = json.dumps(creds).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            body = resp.read().decode("utf-8")
            result = json.loads(body)
            if result.get("success"):
                return True, "凭证已写入服务器"
            else:
                return False, result.get("error", "写入失败")
    except urllib.error.HTTPError as e:
        try:
            body = e.read().decode("utf-8", errors="replace")
            msg = json.loads(body).get("error", str(e))
        except Exception:
            msg = str(e)
        return False, msg
    except Exception as e:
        return False, str(e)


class RelayAuthHandler(http.server.BaseHTTPRequestHandler):
    relay_url = DEFAULT_RELAY_URL
    client_id = DEFAULT_CLIENT_ID
    auth_url = DEFAULT_AUTH_URL

    def log_message(self, fmt, *args):
        pass  # 不输出请求日志

    def _html(self, content: str) -> str:
        return f"""<!doctype html>
<html lang="zh-CN">
<head><meta charset="utf-8"><title>Trae CN Relay 授权</title>
<style>
* {{margin:0;padding:0;box-sizing:border-box}}
body {{font:15px -apple-system,BlinkMacSystemFont,Segoe UI,Roboto,sans-serif;background:#0f1117;color:#e8eaed;display:flex;align-items:center;justify-content:center;min-height:100vh;padding:20px}}
.panel {{background:#1a1d28;border-radius:8px;max-width:460px;width:100%;padding:28px 24px;border:1px solid #2d3140}}
h1 {{font-size:18px;font-weight:600;margin-bottom:12px}}
p {{color:#9aa0b0;font-size:13px;line-height:1.5;margin-bottom:12px}}
.btn {{display:inline-flex;align-items:center;justify-content:center;padding:10px 20px;border-radius:6px;font-size:14px;font-weight:500;cursor:pointer;border:1px solid transparent;text-decoration:none;color:#fff}}
.btn-primary {{background:#1a8c5c;border-color:#1a8c5c}}
.btn-primary:hover {{background:#14a06a}}
.btn-primary:disabled {{opacity:.5;cursor:not-allowed}}
.btn-ghost {{background:transparent;border-color:#3a3f54;color:#9aa0b0}}
.btn-ghost:hover {{border-color:#5a5f74;color:#e8eaed}}
.btn-group {{display:flex;gap:8px;margin-top:12px}}
.msg {{margin-top:12px;padding:8px 12px;border-radius:6px;font-size:13px;display:none}}
.msg-ok {{background:#1f6c3a;color:#a8e6b8;display:block}}
.msg-err {{background:#6c1f1f;color:#e6a8a8;display:block}}
.loading {{margin-top:12px;display:none;font-size:13px;color:#9aa0b0}}
code {{background:#252836;padding:1px 6px;border-radius:4px;font-size:12px}}
</style></head>
<body><div class="panel">{content}</div></body></html>"""

    def _index_page(self) -> str:
        port = self.server.server_address[1]
        relay = self.relay_url
        client_id_js = json.dumps(self.client_id, ensure_ascii=False).replace("<", "\\u003c")
        auth_url_js = json.dumps(self.auth_url, ensure_ascii=False).replace("<", "\\u003c")
        return self._html(f"""
<h1>Trae CN Relay 授权</h1>
<p>
  监听端口: <code>{port}</code><br>
  中转站: <code>{relay}</code>
</p>
<p>
  1. 确保浏览器已登录 <a href="https://www.trae.cn" target="_blank" rel="noopener" style="color:#8ab4f8">trae.cn</a><br>
  2. 点击下方按钮
</p>
<div class="btn-group">
  <button class="btn btn-primary" onclick="startAuth()" id="auth-btn">使用 Trae 网页授权登录</button>
  <a class="btn btn-ghost" href="https://www.trae.cn" target="_blank" rel="noopener">打开 trae.cn</a>
</div>
<div id="loading" class="loading">等待授权中…</div>
<div id="auth-msg" class="msg"></div>
<script>
var state = {{ traceId: null, win: null }};
function uuid(){{return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g,function(c){{var r=Math.random()*16|0;return(c==='x'?r:(r&3|8)).toString(16)}})}}
function randomHex(n){{var a=new Uint8Array(n);crypto.getRandomValues(a);return Array.from(a,b=>b.toString(16).padStart(2,'0')).join('')}}
function randomDigits(n){{var s='';while(s.length<n)s+=Math.floor(Math.random()*1e10).toString();return s.slice(0,n)}}
function buildAuthUrl(){{
  var cb = 'http://127.0.0.1:{port}/authorize';
  var mid = randomHex(32), did = randomDigits(19), tid = state.traceId;
  var p = new URLSearchParams({{
    login_version:'1',auth_from:'solo',login_channel:'native_ide',plugin_version:'2.3.24254',
    auth_type:'local',client_id:{client_id_js},redirect:'0',login_trace_id:tid,
    auth_callback_url:cb,machine_id:mid,device_id:did,x_device_id:did,x_machine_id:mid,
    x_device_brand:'Mac14,7',x_device_type:'mac',x_os_version:'macOS 26.4.1',x_env:'',
    x_app_version:'0.1.7',x_app_type:'stable',hide_saas_login:'true'
  }});
  return {auth_url_js}+'?'+p.toString();
}}
function startAuth(){{
  state.traceId = uuid();
  var url = buildAuthUrl();
  var w = window.open(url, 'trae-relay-oauth', 'width=560,height=760');
  if (!w){{ showMsg('auth-msg','弹出窗口被拦截',false); return; }}
  state.win = w;
  document.getElementById('loading').style.display='block';
  document.getElementById('auth-btn').disabled=true;
  var poll = setInterval(function(){{ if(w.closed){{ clearInterval(poll);document.getElementById('loading').style.display='none';document.getElementById('auth-btn').disabled=false; }} }},700);
}}
window.addEventListener('message',function(ev){{
  if (!ev.data||ev.data.type!=='trae-relay-web-login') return;
  if (state.traceId&&ev.data.loginTraceId!==state.traceId) return;
  if (ev.data.success){{ showMsg('auth-msg','授权成功',true); }}
  else{{ showMsg('auth-msg',ev.data.error||'授权失败',false); }}
  document.getElementById('loading').style.display='none';
  document.getElementById('auth-btn').disabled=false;
  if (state.win&&!state.win.closed) state.win.close();
}});
function showMsg(id,text,ok){{ var el=document.getElementById(id);el.textContent=text;el.className='msg'+(ok?' msg-ok':' msg-err'); }}
</script>""")

    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        params = dict(urllib.parse.parse_qsl(parsed.query))

        if parsed.path == "/healthz":
            self._send_json({"status": "ok", "relay": self.relay_url, "port": self.server.server_address[1]})
        elif parsed.path == "/relay-url":
            self._send_json({"relay": self.relay_url})
        elif parsed.path == "/authorize":
            self._handle_authorize(params)
        elif parsed.path in ("/", "/index.html", ""):
            self._send_html(self._index_page())
        else:
            self.send_response(404)
            self.end_headers()
            self.wfile.write(b"Not found")

    def _send_json(self, obj: dict):
        body = json.dumps(obj).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Access-Control-Max-Age", "86400")
        self.end_headers()

    def _handle_authorize(self, query: dict):
        creds = parse_oauth_params(query)
        trace_id = query.get("loginTraceID") or query.get("login_trace_id") or ""
        if not creds.get("token"):
            html = self._oauth_result_page(False, "未收到有效的 userJwt，请确认已登录 trae.cn", trace_id)
            self._send_html(html)
            return

        # 转发到 relay
        success, msg = forward_to_relay(self.relay_url, creds)
        html = self._oauth_result_page(success, msg, trace_id)
        self._send_html(html)

    def _oauth_result_page(self, success: bool, message: str, login_trace_id: str = "") -> str:
        safe_msg = message.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;").replace('"', "&quot;").replace("'", "&#39;")
        js_message = json.dumps(message, ensure_ascii=False).replace("<", "\\u003c")
        js_trace = json.dumps(login_trace_id, ensure_ascii=False).replace("<", "\\u003c")
        return self._html(f"""
<div class="msg {'msg-ok' if success else 'msg-err'}">
  <h2 style="margin:0 0 8px;font-size:16px">{"成功" if success else "失败"}</h2>
  <p>{safe_msg}</p>
</div>
<script>
(function(){{
  try {{
    if (!window.opener) return;
    var msg = {{ type:'trae-relay-web-login', success:{"true" if success else "false"}, error:null, loginTraceId:{js_trace} }};
    if (!{str(success).lower()}) msg.error = {js_message};
    window.opener.postMessage(msg, '*');
  }} catch(e){{}}
  setTimeout(function(){{ window.close(); }}, {800 if success else 4000});
}})();
</script>
""")

    def _send_html(self, html: str):
        body = html.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def main():
    args = parse_args()
    relay = normalize_relay_url(args.relay)
    if not relay:
        raise SystemExit("Relay 地址无效，请从当前部署重新下载授权助手。")
    if not 1 <= args.port <= 65535:
        raise SystemExit("本机监听端口必须介于 1 和 65535。")
    port = args.port
    client_id = args.client_id
    auth_url = args.auth_url

    handler = RelayAuthHandler
    handler.relay_url = relay
    handler.client_id = client_id
    handler.auth_url = auth_url

    server = http.server.HTTPServer(("127.0.0.1", port), handler)
    print(f"Trae CN Relay 授权回调监听器已启动")
    print(f"  本地监听: http://127.0.0.1:{port}")
    print(f"  中转站:   {relay}")
    print(f"  授权 URL: {auth_url}")
    print(f"  ClientID: {client_id}")
    print()
    print(f"请在浏览器中打开 http://127.0.0.1:{port}")
    print(f"或直接在中转站页面点击授权按钮")
    print()

    if not args.no_open:
        try:
            # Trae 授权完成后回调到本机 /authorize；打开 relay 控制台方便用户点授权。
            webbrowser.open(f"{relay}/web/login")
        except Exception:
            pass

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止")
        sys.exit(0)


if __name__ == "__main__":
    main()