from __future__ import annotations import json import os import ssl import urllib.error import urllib.request import base64 import uuid from datetime import datetime, timezone from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any HOST = "127.0.0.1" PORT = 8787 API_PREFIX = "/api" ROOT = Path(__file__).resolve().parent CHECKOUT_URLS = ( "https://chatgpt.com/backend-api/payments/checkout", "https://chat.openai.com/backend-api/payments/checkout", ) UPSTREAM_PROXY = str(os.getenv("LONG_LINK_UPSTREAM_PROXY") or "").strip() try: from curl_cffi import requests as curl_requests except Exception: curl_requests = None def _json_bytes(payload: dict[str, Any], *, status: int) -> tuple[int, bytes]: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") return status, body def _decode_jwt_payload(token: str) -> dict[str, Any]: try: parts = str(token or "").split(".") if len(parts) != 3: return {} payload = parts[1] payload += "=" * ((4 - len(payload) % 4) % 4) decoded = base64.urlsafe_b64decode(payload.encode("utf-8")).decode("utf-8", errors="replace") data = json.loads(decoded) return data if isinstance(data, dict) else {} except Exception: return {} def _token_diagnostics(token: str) -> dict[str, Any]: payload = _decode_jwt_payload(token) exp = payload.get("exp") now_ts = int(datetime.now(timezone.utc).timestamp()) auth_claim = payload.get("https://api.openai.com/auth") or {} diagnostics = { "token_decoded": bool(payload), "token_expired": bool(isinstance(exp, int) and exp <= now_ts), "token_exp_utc": ( datetime.fromtimestamp(exp, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") if isinstance(exp, int) else "" ), "token_plan_type": str(auth_claim.get("chatgpt_plan_type") or ""), "token_account_id": str(auth_claim.get("chatgpt_account_id") or "")[:12], "proxy_enabled": bool(UPSTREAM_PROXY), } return diagnostics def _pricing_diagnostics(remote_body: bytes, content_type: str) -> dict[str, Any]: text = "" try: text = remote_body.decode("utf-8", errors="replace") except Exception: text = "" diag: dict[str, Any] = { "response_bytes": len(remote_body), "content_type": content_type, "has_free_trial_text": "free trial" in text.lower(), "has_plus_month_free_text": "plus-1-month-free" in text.lower(), "body_preview": text[:300], } if "application/json" in (content_type or "").lower(): try: parsed = json.loads(text) if isinstance(parsed, dict): # 常见结构：{"url":"https://pay.openai.com/c/pay/cs_live_..."} diag["json_keys"] = sorted(list(parsed.keys()))[:20] if "url" in parsed: diag["checkout_url_prefix"] = str(parsed.get("url") or "")[:120] except Exception: pass return diag class LonglinkHandler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def end_headers(self) -> None: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS") super().end_headers() def do_OPTIONS(self) -> None: # noqa: N802 self.send_response(204) self.end_headers() def do_GET(self) -> None: # noqa: N802 if self.path.rstrip("/") == f"{API_PREFIX}/health": status, body = _json_bytes( { "ok": True, "service": "longlink_server", "proxy_enabled": bool(UPSTREAM_PROXY), }, status=200, ) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return super().do_GET() def do_POST(self) -> None: # noqa: N802 normalized_path = self.path.rstrip("/") if normalized_path not in {f"{API_PREFIX}/request", f"{API_PREFIX}/checkout"}: status, body = _json_bytes({"error": "not_found"}, status=404) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return content_length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(content_length) if content_length > 0 else b"" try: data = json.loads(raw.decode("utf-8")) except Exception: status, body = _json_bytes({"error": "invalid_json"}, status=400) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return token = str(data.get("token") or "").strip() if token.lower().startswith("bearer "): token = token[7:].strip() if not token: status, body = _json_bytes({"error": "missing_token"}, status=400) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return checkout_payload = { "plan_name": str(data.get("plan_name") or "chatgptplusplan"), "billing_details": data.get("billing_details") or {"country": "DE", "currency": "EUR"}, "cancel_url": str(data.get("cancel_url") or "https://chat.openai.com/#pricing"), "promo_campaign": data.get("promo_campaign") or {"promo_campaign_id": "plus-1-month-free", "is_coupon_from_query_param": False}, "checkout_ui_mode": str(data.get("checkout_ui_mode") or "hosted"), } request_id = uuid.uuid4().hex[:10] print( json.dumps( { "tag": "checkout_request", "request_id": request_id, "billing_details": checkout_payload.get("billing_details"), "plan_name": checkout_payload.get("plan_name"), "promo_campaign": checkout_payload.get("promo_campaign"), "proxy_enabled": bool(UPSTREAM_PROXY), }, ensure_ascii=False, ), flush=True, ) status_code, content_type, remote_body, err, forward_meta = self._forward_checkout(token, checkout_payload) if err: print( json.dumps( { "tag": "checkout_forward_error", "request_id": request_id, "error": err, "forward_meta": forward_meta, }, ensure_ascii=False, ), flush=True, ) status, body = _json_bytes({"error": err}, status=502) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return print( json.dumps( { "tag": "checkout_response", "request_id": request_id, "status_code": status_code, "forward_meta": forward_meta, "pricing_diag": _pricing_diagnostics(remote_body, content_type), }, ensure_ascii=False, ), flush=True, ) if status_code >= 400: preview = "" try: preview = remote_body.decode("utf-8", errors="replace")[:300] except Exception: preview = "" print(f"[longlink] upstream HTTP {status_code}, body preview: {preview}", flush=True) if status_code == 403 and (not preview.strip() or preview.strip() in {"{}", "null"}): diag = _token_diagnostics(token) status, body = _json_bytes( { "error": "upstream_forbidden", "detail": "Upstream rejected this request (likely risk-control / IP reputation / account restriction).", "diagnostics": diag, }, status=403, ) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return try: self.send_response(status_code) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(remote_body))) self.end_headers() self.wfile.write(remote_body) except Exception as exc: status, body = _json_bytes({"error": f"proxy_response_exception: {exc}"}, status=502) self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _forward_checkout( self, token: str, checkout_payload: dict[str, Any] ) -> tuple[int, str, bytes, str, dict[str, Any]]: attempts: list[dict[str, Any]] = [] if curl_requests is not None: try: errors: list[str] = [] for checkout_url in CHECKOUT_URLS: response = curl_requests.post( checkout_url, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", "Origin": "https://chatgpt.com", "Referer": "https://chatgpt.com/", "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36" ), }, json=checkout_payload, timeout=30, impersonate="chrome", proxy=UPSTREAM_PROXY or None, ) body = bytes(response.content or b"") status = int(response.status_code or 500) content_type = str(response.headers.get("Content-Type") or "application/json; charset=utf-8") attempts.append({"url": checkout_url, "transport": "curl_cffi", "status": status}) # 2xx / 4xx 直接返回，让前端看到远端真实响应；5xx 才尝试下一个域名。 if status < 500: return status, content_type, body, "", { "selected_url": checkout_url, "transport": "curl_cffi", "attempts": attempts, } errors.append(f"{checkout_url} -> HTTP {status}") return 0, "application/json; charset=utf-8", b"", ( f"proxy_exception: all_checkout_urls_failed [{'; '.join(errors)}]" ), {"attempts": attempts} except Exception as exc: # curl_cffi path failed, fallback to urllib below. curl_err = f"curl_cffi_exception: {exc!r}" attempts.append({"transport": "curl_cffi", "error": curl_err}) else: curl_err = "" else: curl_err = "curl_cffi_not_available" opener = None if UPSTREAM_PROXY: opener = urllib.request.build_opener( urllib.request.ProxyHandler({"http": UPSTREAM_PROXY, "https": UPSTREAM_PROXY}) ) ctx = ssl.create_default_context() fallback_errors: list[str] = [] for checkout_url in CHECKOUT_URLS: req_data = json.dumps(checkout_payload).encode("utf-8") req = urllib.request.Request( checkout_url, data=req_data, method="POST", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", "Origin": "https://chatgpt.com", "Referer": "https://chatgpt.com/", "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36" ), }, ) try: if opener is not None: resp = opener.open(req, timeout=30) else: resp = urllib.request.urlopen(req, timeout=30, context=ctx) with resp: remote_body = resp.read() status_code = int(resp.getcode() or 200) content_type = resp.headers.get("Content-Type", "application/json; charset=utf-8") attempts.append({"url": checkout_url, "transport": "urllib", "status": status_code}) if status_code < 500: return status_code, content_type, remote_body, "", { "selected_url": checkout_url, "transport": "urllib", "attempts": attempts, } fallback_errors.append(f"{checkout_url} -> HTTP {status_code}") except urllib.error.HTTPError as exc: remote_body = exc.read() or b"" status_code = int(exc.code or 500) headers = exc.headers or {} content_type = headers.get("Content-Type", "application/json; charset=utf-8") attempts.append({"url": checkout_url, "transport": "urllib", "status": status_code}) if status_code < 500: return status_code, content_type, remote_body, "", { "selected_url": checkout_url, "transport": "urllib", "attempts": attempts, } fallback_errors.append(f"{checkout_url} -> HTTP {status_code}") except Exception as exc: attempts.append({"url": checkout_url, "transport": "urllib", "error": repr(exc)}) fallback_errors.append(f"{checkout_url} -> {exc!r}") joined = "; ".join(fallback_errors) if fallback_errors else "unknown_error" return 0, "application/json; charset=utf-8", b"", f"proxy_exception: {joined} [{curl_err}]", { "attempts": attempts } def main() -> None: server = ThreadingHTTPServer((HOST, PORT), LonglinkHandler) print(f"[longlink] serving at http://{HOST}:{PORT}/longlink.html") print(f"[longlink] health check: http://{HOST}:{PORT}{API_PREFIX}/health") print(f"[longlink] upstream proxy: {'enabled' if UPSTREAM_PROXY else 'disabled'}") server.serve_forever() if __name__ == "__main__": main()