Logic/gptplus_machine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add register-only and checkout commands, fix payment flow, switch to yyds mail

Browse Source 
- add 'register' command: register account and print credentials only
- add 'checkout' command: register account and print Plus hosted checkout URL
- fix hCaptcha rqdata/isInvisible for Stripe setup intent verification
- fix verify_challenge flow: call chatgpt.com/checkout/verify after challenge
- fix currency uppercase for checkout API
- add get_checkout_url method to ChatGPTPayment
- switch default mail provider to yyds
- update README with new commands and usage
This commit is contained in:
Logic
2026-03-21 14:38:28 +08:00
parent 3ca69eb5d3
commit 15cba50314
8 changed files with 682 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
README.md
View File

@@ -1,17 +1,18 @@
# gptplus_auto
# gptplus_machine
ChatGPT Plus 自动注册、订阅工具，以及 Codex CLI OAuth 登录工具。
ChatGPT 账号自动注册工具，以及 Codex CLI OAuth 登录工具。
## 功能
- **register** — 自动注册新 ChatGPT 账号，并可选完成 Plus 订阅（Stripe 支付）
- **codex-login** — 对已有 ChatGPT 账号执行 Codex CLI OAuth 登录，纯 HTTP 实现，无需浏览器
- **register** — 自动注册新 ChatGPT 账号，输出邮箱、密码、邮箱凭证
- **checkout** — 注册账号后生成 Plus 支付链接（首月免费），在浏览器完成支付即可开通
- **codex-login** — 对已有账号执行 Codex CLI OAuth 登录，纯 HTTP 实现，无需浏览器
## 安装
```bash
python -m venv .venv
source .venv/bin/activate
uv sync
# 或
pip install -r requirements.txt
```
@@ -23,73 +24,92 @@ pip install -r requirements.txt
# 代理（推荐美国 IP）
SOCKS5_PROXY=socks5://user:pass@host:port
# 临时邮箱服务（vmail 或 mailtm）
MAIL_PROVIDER=vmail
# YesCaptcha（注册功能需要）
YESCAPTCHA_API_KEY=your_key_here
# 支付信息（订阅功能需要，可留空跳过支付）
CARD_NUMBER=4111111111111111
CARD_EXP_MONTH=12
CARD_EXP_YEAR=2028
CARD_CVC=123
BILLING_NAME=John Smith
BILLING_EMAIL=john@example.com
BILLING_ADDRESS_LINE1=123 Main St
BILLING_ADDRESS_CITY=New York
BILLING_ADDRESS_STATE=NY
BILLING_ADDRESS_POSTAL_CODE=10001
# 支付地区（checkout 命令使用）
COUNTRY=US
CURRENCY=usd
```
## 使用
所有功能通过 `src/main.py` 统一入口调用：
### 注册账号（+ 可选 Plus 订阅）
### 仅注册账号
```bash
.venv/bin/python src/main.py register
uv run python src/main.py register
```
- 自动申请临时邮箱、注册账号、解 hCaptcha
- 若 `.env` 中配置了信用卡信息，注册完成后自动订阅 ChatGPT Plus
- 输出邮箱、邮箱密码、ChatGPT 密码、access token
输出示例：
```
=== Account Created ===
email: abc123@vmail.dev
password: Xk9#mPqLwZ2!vBnR
mailbox_id: aBcDeFgHiJkLmNoPq
mailbox_password: (空则无需密码)
access_token: eyJhbGci...
```
### 注册账号 + 获取 Plus 支付链接
```bash
uv run python src/main.py checkout
```
输出示例：
```
=== Account Created ===
email: abc123@vmail.dev
password: Xk9#mPqLwZ2!vBnR
mailbox_id: aBcDeFgHiJkLmNoPq
=== Plus Checkout URL ===
https://pay.openai.com/c/pay/cs_live_a1...
```
在浏览器打开链接，填入信用卡信息完成支付（新账号享首月免费优惠），Plus 即开通。
### Codex CLI OAuth 登录
```bash
.venv/bin/python src/main.py codex-login --email user@example.com --password yourpassword
uv run python src/main.py codex-login --email user@example.com --password yourpassword
```
可选参数：
| 参数 | 说明 |
|------|------|
| `--email` | 账号邮箱（不传则交互式输入）|
| `--password` | 账号密码（不传则交互式输入）|
| `--email` | 账号邮箱 |
| `--password` | 账号密码 |
| `--otp` | 邮箱 OTP（如需要）|
| `--workspace-id` | 指定 workspace ID |
| `--authorize-url` | 自定义 OAuth authorize URL |
| `--mailbox-id` | vmail.dev mailbox ID（自动获取 OTP）|
| `--mailbox-password` | vmail.dev mailbox 密码 |
成功后输出 `localhost:1455/auth/callback?code=...` 回调 URL，可直接交给 Codex CLI 完成登录。
成功后输出 `localhost:1455/auth/callback?code=...` 回调 URL，交给 Codex CLI 完成登录。
## 项目结构
```
src/
├── main.py # 统一入口（register / codex-login）
├── main.py # 入口（register / checkout / codex-login）
├── config.py # 配置（pydantic-settings，读取 .env）
├── http_client.py # HTTP 客户端（curl_cffi Chrome 模拟）
├── vmail_client.py # 临时邮箱（mail.tm）
├── captcha_solver.py # YesCaptcha hCaptcha 解决
├── chatgpt_register_http_reverse.py # 完整注册流程
├── chatgpt_payment.py # 完整 Stripe 支付流程
└── codex_oauth_http_flow.py # Codex CLI OAuth 登录（纯 HTTP）
├── vmail_client.py # 临时邮箱客户端
├── captcha_solver.py # YesCaptcha hCaptcha 解题
├── chatgpt_register_http_reverse.py # 注册流程
├── chatgpt_payment.py # Stripe checkout 流程
└── codex_oauth_http_flow.py # Codex CLI OAuth 登录
```
## 注意事项
1. **仅供学习研究**：请遵守 OpenAI 服务条款
2. **代理建议**：建议使用美国 IP 代理，避免触发风控
3. **API 可能变化**：OpenAI/Stripe 可能随时更改接口
4. **避免频繁调用**：同 IP 短时间内多次注册可能被封
- 建议使用美国 IP 代理，避免触发风控
- 同 IP 短时间内多次注册可能被封，建议间隔使用
- checkout 链接有时效（约30分钟），生成后尽快使用

16
src/captcha_solver.py
View File

@@ -9,18 +9,22 @@ class CaptchaSolver:
def __init__(self, api_key: str):
self.api_key = api_key
def solve_hcaptcha(self, site_key: str, page_url: str, timeout: int = 180) -> str:
def solve_hcaptcha(self, site_key: str, page_url: str, timeout: int = 180, rqdata: str = "") -> str:
"""Submit hCaptcha task via YesCaptcha and wait for solution token."""
task = {
"type": "HCaptchaTaskProxyless",
"websiteURL": page_url,
"websiteKey": site_key,
}
if rqdata:
task["isInvisible"] = True
task["enterprisePayload"] = {"rqdata": rqdata}
# createTask
r = httpx.post(
f"{YESCAPTCHA_BASE}/createTask",
json={
"clientKey": self.api_key,
"task": {
"type": "HCaptchaTaskProxyless",
"websiteURL": page_url,
"websiteKey": site_key,
},
"task": task,
},
timeout=30,
)

234
src/chatgpt_payment.py
View File

@@ -10,9 +10,55 @@ class ChatGPTPayment:
self.captcha = captcha
self.base = "https://chatgpt.com"
def subscribe_plus(
self, access_token: str, country: str, currency: str, card_info: dict
) -> bool:
def get_checkout_url(self, access_token: str, country: str = "US", currency: str = "usd") -> str:
"""Create a Stripe hosted checkout session and return the payment URL."""
# Fetch promo campaign
promo_campaign_id = ""
plan_name = "chatgptplusplan"
r = self.http.request(
"GET",
f"{self.base}/backend-api/accounts/check/v4-2023-04-27",
headers={"Authorization": f"Bearer {access_token}"},
)
if r.status_code == 200:
accounts = r.json().get("accounts", {})
for acct in accounts.values():
plus_promo = acct.get("eligible_promo_campaigns", {}).get("plus")
if plus_promo:
promo_campaign_id = plus_promo.get("id", "")
plan_name = plus_promo.get("metadata", {}).get("plan_name", "chatgptplusplan")
print(f" Promo found: {promo_campaign_id} plan={plan_name}")
break
payload = {
"plan_name": plan_name,
"billing_details": {"country": country, "currency": currency.upper()},
"prefetch": True,
"checkout_ui_mode": "hosted",
}
if promo_campaign_id:
payload["promo_campaign"] = {"promo_campaign_id": promo_campaign_id, "is_coupon_from_query_param": False}
r = self.http.request(
"POST",
f"{self.base}/backend-api/payments/checkout",
json=payload,
headers={"Authorization": f"Bearer {access_token}"},
)
checkout = r.json()
print(f" Checkout response keys: {list(checkout.keys())}")
print(f" Checkout response: {checkout}")
url = checkout.get("stripe_hosted_url") or checkout.get("url") or checkout.get("checkout_url")
if not url:
# Try custom mode and extract from session
session_id = checkout.get("checkout_session_id")
if session_id:
url = f"https://checkout.stripe.com/c/pay/{session_id}"
if not url:
raise RuntimeError(f"Could not get checkout URL: {checkout}")
return url
"""Subscribe to ChatGPT Plus."""
checkout_currency = currency.upper()
@@ -23,6 +69,10 @@ class ChatGPTPayment:
f"{self.base}/backend-api/accounts/check/v4-2023-04-27",
headers={"Authorization": f"Bearer {access_token}"},
)
if r.status_code == 403:
raise RuntimeError(f"Access token expired or invalid (403). Token may have expired. Status: {r.status_code}")
if r.status_code != 200:
print(f" Warning: Account check returned {r.status_code}, trying anyway...")
accounts_data = r.json()
# Extract promo campaign from first account
@@ -42,19 +92,42 @@ class ChatGPTPayment:
# 1. Create checkout session
print("[1/6] Creating checkout session...")
checkout_payload = {
"plan_name": plan_name,
"billing_details": {"country": country, "currency": checkout_currency},
"prefetch": True,
"checkout_ui_mode": "custom",
}
# Only include promo_campaign if we have a valid promo_campaign_id (not empty string)
if promo_campaign_id:
checkout_payload["promo_campaign"] = {"promo_campaign_id": promo_campaign_id, "is_coupon_from_query_param": False}
r = self.http.request(
"POST",
f"{self.base}/backend-api/payments/checkout",
json=checkout_payload,
headers={"Authorization": f"Bearer {access_token}"},
)
checkout = r.json()
if "checkout_session_id" not in checkout:
print(f" Warning: checkout response missing checkout_session_id: {checkout}")
# Try without promo campaign
print(" Retrying without promo campaign...")
r = self.http.request(
"POST",
f"{self.base}/backend-api/payments/checkout",
json={
"plan_name": plan_name,
"billing_details": {"country": country, "currency": checkout_currency},
"promo_campaign": {"promo_campaign_id": promo_campaign_id, "is_coupon_from_query_param": False},
"prefetch": True,
"checkout_ui_mode": "custom",
"checkout_ui_mode": "hosted",
},
headers={"Authorization": f"Bearer {access_token}"},
)
checkout = r.json()
if "checkout_session_id" not in checkout:
raise RuntimeError(f"Checkout still failed: {checkout}")
session_id = checkout["checkout_session_id"]
publishable_key = checkout["publishable_key"]
processor_entity = checkout["processor_entity"]
@@ -220,11 +293,156 @@ class ChatGPTPayment:
raise RuntimeError(f"Stripe confirm failed: {r.text}")
print(f" Stripe confirm status: {r.status_code}")
# 6. Verify subscription
# Check Stripe confirm response for errors
confirm_data = r.json()
stripe_status = confirm_data.get("status", "unknown")
payment_status = confirm_data.get("payment_status", "unknown")
print(f" Stripe session status: {stripe_status}")
print(f" Stripe payment status: {payment_status}")
# Handle requires_action - this means Stripe needs additional verification (3DS, CAPTCHA, etc.)
if stripe_status == "requires_action":
next_action = confirm_data.get("next_action") or {}
action_type = next_action.get("type") if isinstance(next_action, dict) else None
print(f" Stripe requires action: {action_type}")
if action_type == "use_stripe_sdk":
stripe_js = next_action.get("use_stripe_sdk") or {}
site_key = stripe_js.get("site_key")
verification_url = stripe_js.get("verification_url")
print(f" Site key: {site_key}")
print(f" Verification URL: {verification_url}")
raise RuntimeError(f"Payment requires additional verification (3DS/CAPTCHA). Site key: {site_key}")
# Check for payment intent errors
payment_intent = confirm_data.get("payment_intent")
if payment_intent and isinstance(payment_intent, dict):
pi_status = payment_intent.get("status", "unknown")
print(f" Payment intent status: {pi_status}")
if pi_status == "requires_payment_method":
error_msg = payment_intent.get("last_payment_error", {}).get("message", "unknown") if isinstance(payment_intent.get("last_payment_error"), dict) else "unknown"
raise RuntimeError(f"Payment failed: {error_msg}")
# If payment_status is "paid" or session status is "complete", payment succeeded
if payment_status == "paid" or stripe_status == "complete":
print(" Payment confirmed as successful!")
elif stripe_status == "open" and payment_status == "unpaid":
# Check if there's a setup intent requiring action
setup_intent = confirm_data.get("setup_intent")
if setup_intent and isinstance(setup_intent, dict):
si_status = setup_intent.get("status")
print(f" Setup intent status: {si_status}")
if si_status == "requires_action":
next_action = setup_intent.get("next_action") or {}
action_type = next_action.get("type") if isinstance(next_action, dict) else None
print(f" Setup requires action: {action_type}")
if action_type == "use_stripe_sdk":
stripe_sdk = next_action.get("use_stripe_sdk") or {}
stripe_js = stripe_sdk.get("stripe_js") or {}
site_key = stripe_js.get("site_key")
verification_url = stripe_js.get("verification_url")
client_secret = setup_intent.get("client_secret", "")
rqdata = stripe_js.get("rqdata", "")
print(f" Second hCaptcha site key: {site_key}")
print(f" Verification URL: {verification_url}")
print(f" rqdata present: {bool(rqdata)}")
if not site_key:
raise RuntimeError("setup_intent requires_action but no site_key found")
# Solve the second hCaptcha - host must be b.stripecdn.com
print(" Solving second hCaptcha for setup intent...")
setup_captcha_token = self.captcha.solve_hcaptcha(site_key, "https://b.stripecdn.com", rqdata=rqdata)
print(f" Setup captcha token: {setup_captcha_token[:30]}...")
# Submit verification to Stripe:
# URL: https://api.stripe.com + verification_url
# Body: challenge_response_token + client_secret + key
if verification_url:
verify_url = f"https://api.stripe.com{verification_url}"
print(f" Submitting setup intent verification to {verify_url}")
verify_r = self.http.request(
"POST",
verify_url,
data={
"challenge_response_token": setup_captcha_token,
"challenge_response_ekey": "",
"client_secret": client_secret,
"captcha_vendor_name": "hcaptcha",
"key": publishable_key,
"_stripe_version": "2025-03-31.basil; checkout_server_update_beta=v1; checkout_manual_approval_preview=v1",
},
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Origin": "https://js.stripe.com",
"Referer": "https://js.stripe.com/",
},
skip_pacing=True,
)
print(f" Setup verification status: {verify_r.status_code}")
if verify_r.status_code == 200:
verify_result = verify_r.json()
print(f" Verification result: {verify_result}")
# After verify_challenge, call chatgpt.com/checkout/verify to finalize subscription
chatgpt_verify_url = f"https://chatgpt.com/checkout/verify?stripe_session_id={session_id}&processor_entity={processor_entity}&plan_type=plus"
print(f" Calling ChatGPT verify URL...")
verify_chatgpt_r = self.http.request(
"GET",
chatgpt_verify_url,
headers={"Authorization": f"Bearer {access_token}"},
skip_pacing=True,
)
print(f" ChatGPT verify status: {verify_chatgpt_r.status_code}")
if verify_chatgpt_r.status_code == 200:
payment_status = "paid"
stripe_status = "complete"
else:
print(f" Setup verification failed: {verify_r.text[:200]}")
raise RuntimeError("Setup intent verification failed")
else:
raise RuntimeError(f"Payment requires additional verification. Action: {action_type}")
# 6. Verify subscription by checking account status
print("[6/6] Verifying subscription...")
r = self.http.request(
"GET",
f"{self.base}/backend-api/payments/checkout/{processor_entity}/{session_id}",
f"{self.base}/backend-api/accounts/check/v4-2023-04-27",
headers={"Authorization": f"Bearer {access_token}"},
)
return r.status_code == 200
print(f" Account check status: {r.status_code}")
print(f" Account check response: {r.text[:2000]}")
if r.status_code != 200:
print(f" Account check failed: {r.status_code} - {r.text[:200]}")
return False
accounts_data = r.json()
accounts = accounts_data.get("accounts", {})
if not accounts:
print(f" No accounts found")
return False
# Check each account for Plus subscription
has_plus = False
for account_id, account in accounts.items():
subscription = account.get("subscription", {})
if subscription:
plan_id = subscription.get("plan_id", "")
status = subscription.get("status", "")
print(f" Account {account_id}: plan={plan_id}, status={status}")
if "plus" in plan_id.lower() and status == "active":
has_plus = True
if has_plus:
print(" Plus subscription verified!")
return True
# Try alternative check - check for plus entitlements
for account_id, account in accounts.items():
entitlements = account.get("entitlements", [])
for ent in entitlements:
if ent.get("feature") == "plus" and ent.get("status") == "active":
print(f" Plus entitlement found in account {account_id}")
has_plus = True
if not has_plus:
print(f" Plus subscription NOT found. Full response: {accounts_data}")
return has_plus

4
src/chatgpt_register_http_reverse.py
View File

@@ -6,7 +6,8 @@ from typing import Optional
from urllib.parse import unquote, urlencode, urljoin, urlparse
from http_client import HTTPClient
from vmail_client import MailClient
from vmail_client import get_mail_client
from config import settings
class ChatGPTRegisterHTTPReverse:
@@ -411,6 +412,7 @@ class ChatGPTRegisterHTTPReverse:
print(f" Session status: {session_response.status_code}")
return {
"email": email,
"mailbox_id": mailbox.get("id", ""),
"mailbox_password": mailbox.get("password", ""),
"access_token": session_data.get("accessToken", ""),
"user_id": session_data.get("user", {}).get("id", ""),

8
src/codex_oauth_http_flow.py
View File

@@ -199,6 +199,7 @@ class CodexOAuthHTTPFlow:
workspace_id: str = "",
use_browser: bool = False,
mailbox: dict | None = None,
mail_client = None,
) -> None:
self.authorize_url = authorize_url
self.email = email
@@ -206,6 +207,7 @@ class CodexOAuthHTTPFlow:
self.otp = otp
self.workspace_id = workspace_id
self.mailbox = mailbox
self.mail_client = mail_client
self.auth_base = "https://auth.openai.com"
self.proxy = load_proxy()
self.http = HTTPClient(self.proxy)
@@ -310,8 +312,10 @@ class CodexOAuthHTTPFlow:
"""Try to fetch OTP from mailbox automatically."""
if not self.mailbox:
return ""
from vmail_client import MailClient
mail = MailClient()
mail = self.mail_client
if not mail:
from vmail_client import get_mail_client
mail = get_mail_client()
print("[otp] auto-fetching OTP from mailbox...")
try:
code = mail.wait_for_otp(self.mailbox, timeout=120, poll=3.0)

1
src/config.py
View File

@@ -6,6 +6,7 @@ class Settings(BaseSettings):
vmail_api_key: str = Field(default="", env="VMAIL_API_KEY")
yescaptcha_api_key: str = Field(default="", env="YESCAPTCHA_API_KEY")
socks5_proxy: str = Field(default="", env="SOCKS5_PROXY")
mail_provider: str = Field(default="vmail", env="MAIL_PROVIDER") # "vmail", "mailtm", or "yyds"
# Payment info
card_number: str = Field(default="", env="CARD_NUMBER")

201
src/main.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
"""Entry point: ChatGPT Plus auto-registration + subscription, or Codex OAuth login."""
"""Entry point: ChatGPT account registration, Plus checkout, and Codex OAuth login."""
import argparse
import random
import string
import sys
from config import settings
from vmail_client import MailClient
from vmail_client import get_mail_client
from captcha_solver import CaptchaSolver
from http_client import HTTPClient
from chatgpt_register_http_reverse import ChatGPTRegisterHTTPReverse
@@ -29,152 +29,137 @@ def generate_name():
return f"{first} {last}"
def cmd_register(args):
"""Register a new ChatGPT account and optionally subscribe to Plus."""
def cmd_register_only(args):
"""Register a new ChatGPT account and print credentials."""
password = generate_password()
name = generate_name()
print("Generated credentials:")
print(f" Password: {password}")
print(f" Name: {name}")
mail = MailClient()
mail = get_mail_client(settings.mail_provider)
http = HTTPClient(proxy=settings.socks5_proxy)
http.enable_pacing(min_delay=0.5, max_delay=2.0)
http.enable_pacing(min_delay=0.5, max_delay=3.0)
try:
print("\n=== ChatGPT Registration ===")
register = ChatGPTRegisterHTTPReverse(http, mail)
session = register.register(password, name)
print("\n Registration complete!")
print(f" Email: {session['email']}")
print(f" Mailbox password: {session['mailbox_password']}")
print(f" ChatGPT password: {password}")
if session["access_token"]:
print(f" Access Token: {session['access_token'][:50]}...")
if settings.card_number and session.get("access_token"):
print("\n=== ChatGPT Plus Subscription ===")
captcha = CaptchaSolver(settings.yescaptcha_api_key)
payment = ChatGPTPayment(http, captcha)
success = payment.subscribe_plus(
access_token=session["access_token"],
country=settings.country,
currency=settings.currency,
card_info={
"number": settings.card_number,
"exp_month": settings.card_exp_month,
"exp_year": settings.card_exp_year,
"cvc": settings.card_cvc,
"billing_name": settings.billing_name or name,
"billing_email": settings.billing_email or session["email"],
"billing_address_line1": settings.billing_address_line1,
"billing_address_city": settings.billing_address_city,
"billing_address_state": settings.billing_address_state,
"billing_address_postal_code": settings.billing_address_postal_code,
},
)
if success:
print("\n Plus subscription complete!")
print(f" Email: {session['email']}")
print(f" Mailbox password: {session['mailbox_password']}")
print(f" ChatGPT password: {password}")
else:
print("\n No card info configured, skipping Plus subscription")
print("\n=== Account Created ===")
print(f"email: {session['email']}")
print(f"password: {password}")
if session.get("mailbox_id"):
print(f"mailbox_id: {session['mailbox_id']}")
if session.get("mailbox_password"):
print(f"mailbox_password: {session['mailbox_password']}")
if session.get("access_token"):
print(f"access_token: {session['access_token']}")
return 0
except Exception as e:
print(f"\n Error: {e}")
print(f"\nError: {e}", file=sys.stderr)
import traceback
traceback.print_exc()
return 1
finally:
http.close()
return 0
def cmd_codex_login(args):
"""Run Codex OAuth login and print the callback URL with authorization code."""
email = args.email
password = args.password
if not email:
email = input("Email: ").strip()
if not password:
import getpass
password = getpass.getpass("Password: ")
def cmd_checkout(args):
"""Register a new account and print a Plus checkout URL."""
password = generate_password()
name = generate_name()
authorize_url = args.authorize_url or DEFAULT_AUTHORIZE_URL
mail = get_mail_client(settings.mail_provider)
http = HTTPClient(proxy=settings.socks5_proxy)
http.enable_pacing(min_delay=0.5, max_delay=3.0)
# Try to get mailbox token for auto OTP fetching
mailbox = None
if args.mailbox_password:
import httpx
try:
token_resp = httpx.post(
"https://api.mail.tm/token",
json={"address": email, "password": args.mailbox_password},
timeout=30,
)
if token_resp.status_code == 200:
mailbox = {
"address": email,
"password": args.mailbox_password,
"token": token_resp.json()["token"],
}
print(f"Mailbox token obtained, will auto-fetch OTP if needed")
except Exception as e:
print(f"Warning: Failed to get mailbox token: {e}")
register = ChatGPTRegisterHTTPReverse(http, mail)
session = register.register(password, name)
print(f"Starting Codex OAuth flow for {email}")
flow = CodexOAuthHTTPFlow(
authorize_url=authorize_url,
email=email,
password=password,
otp=args.otp or "",
workspace_id=args.workspace_id or "",
mailbox=mailbox,
http.disable_pacing()
captcha = CaptchaSolver(settings.yescaptcha_api_key)
payment = ChatGPTPayment(http, captcha)
checkout_url = payment.get_checkout_url(
access_token=session["access_token"],
country=settings.country,
currency=settings.currency,
)
try:
callback_url = flow.run()
print("\n" + "=" * 60)
print("[SUCCESS] callback_url:")
print(callback_url)
print("=" * 60)
print("\n=== Account Created ===")
print(f"email: {session['email']}")
print(f"password: {password}")
if session.get("mailbox_id"):
print(f"mailbox_id: {session['mailbox_id']}")
if session.get("mailbox_password"):
print(f"mailbox_password: {session['mailbox_password']}")
print(f"\n=== Plus Checkout URL ===")
print(checkout_url)
return 0
except Exception as e:
print(f"\n[ERROR] {e}")
print(f"\nError: {e}", file=sys.stderr)
import traceback
traceback.print_exc()
return 1
finally:
flow.close()
http.close()
def cmd_codex_login(args):
"""Codex CLI OAuth login via pure HTTP."""
http = HTTPClient(proxy=settings.socks5_proxy)
http.enable_pacing(min_delay=0.5, max_delay=2.0)
email = args.email or input("Email: ").strip()
password = args.password or input("Password: ").strip()
authorize_url = args.authorize_url or DEFAULT_AUTHORIZE_URL
mail = None
if args.mailbox_id:
mail = get_mail_client(settings.mail_provider)
try:
flow = CodexOAuthHTTPFlow(http, mail)
callback_url = flow.login(
email=email,
password=password,
otp=args.otp,
workspace_id=args.workspace_id,
authorize_url=authorize_url,
mailbox_id=args.mailbox_id,
mailbox_password=args.mailbox_password,
)
print(f"\nCallback URL:\n{callback_url}")
return 0
except Exception as e:
print(f"\nError: {e}", file=sys.stderr)
import traceback
traceback.print_exc()
return 1
finally:
http.close()
def build_parser():
parser = argparse.ArgumentParser(
description="ChatGPT Plus auto-registration + subscription, or Codex OAuth login",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Commands:
register Auto-register a new ChatGPT account and subscribe to Plus
codex-login Run Codex CLI OAuth login and obtain the callback URL
Examples:
python src/main.py register
python src/main.py codex-login --email user@example.com --password mypass
"""
prog="main.py",
description="ChatGPT account tools",
)
sub = parser.add_subparsers(dest="command")
# register sub-command
sub.add_parser("register", help="Register new ChatGPT account + optional Plus subscription")
# register-only
sub.add_parser("register", help="Register a new ChatGPT account and print credentials")
# codex-login sub-command
# checkout
sub.add_parser("checkout", help="Register a new account and print a Plus checkout URL")
# codex-login
p_codex = sub.add_parser("codex-login", help="Codex CLI OAuth login (HTTP-only, no browser)")
p_codex.add_argument("--email", default="", help="OpenAI account email")
p_codex.add_argument("--password", default="", help="OpenAI account password")
p_codex.add_argument("--otp", default="", help="Email OTP code (if already known)")
p_codex.add_argument("--workspace-id", dest="workspace_id", default="", help="Workspace ID to select")
p_codex.add_argument("--authorize-url", dest="authorize_url", default="", help="Custom authorize URL")
p_codex.add_argument("--mailbox-password", dest="mailbox_password", default="", help="Mailbox password for auto OTP fetching")
p_codex.add_argument("--mailbox-password", dest="mailbox_password", default="", help="vmail.dev mailbox password for auto OTP")
p_codex.add_argument("--mailbox-id", dest="mailbox_id", default="", help="vmail.dev mailbox ID for auto OTP")
return parser
@@ -184,7 +169,9 @@ def main():
args = parser.parse_args()
if args.command == "register":
return cmd_register(args)
return cmd_register_only(args)
elif args.command == "checkout":
return cmd_checkout(args)
elif args.command == "codex-login":
return cmd_codex_login(args)
else:

286
src/vmail_client.py
View File

@@ -1,18 +1,157 @@
import re
import secrets
import time
from abc import ABC, abstractmethod
import httpx
# vmail.dev config
VMAIL_BASE = "https://vmail.dev/api/v1"
VMAIL_API_KEY = "vmail_UCjaCMZghzKaH5KrRlKXyxkrCO7ZNesV"
# mail.tm config
MAILTM_BASE = "https://api.mail.tm"
# yyds mail config
YYDS_BASE = "https://maliapi.215.im"
YYDS_API_KEY = "AC-4a23c58b8f84c19a27ef509e"
class MailClient:
"""mail.tm temporary email client."""
class BaseMailClient(ABC):
"""Abstract base class for temporary email clients."""
@abstractmethod
def create_mailbox(self) -> dict:
"""Create a mailbox. Returns dict with id, address, password, token."""
pass
@abstractmethod
def wait_for_otp(self, mailbox: dict, timeout: int = 120, poll: float = 5.0) -> str:
"""Poll mailbox until a 6-digit OTP arrives."""
pass
def get_token_for_address(self, address: str, password: str) -> str:
"""Get auth token for an existing mailbox (for mail.tm only)."""
return ""
class VMailClient(BaseMailClient):
"""vmail.dev temporary email client."""
def __init__(self):
self.api_key = VMAIL_API_KEY
self.headers = {"X-API-Key": self.api_key}
self.base = VMAIL_BASE
def create_mailbox(self) -> dict:
"""Create a mailbox. Returns dict with address, password, token."""
domains_resp = httpx.get(f"{MAILTM_BASE}/domains", timeout=30)
"""Create a mailbox. Returns dict with id, address, password, token."""
for attempt in range(10):
local_part = secrets.token_hex(4)
r = httpx.post(
f"{self.base}/mailboxes",
headers=self.headers,
json={"localPart": local_part},
timeout=30,
)
if r.status_code == 201:
data = r.json()["data"]
return {
"id": data["id"],
"address": data["address"],
"password": data.get("password", ""),
"token": self.api_key,
}
if r.status_code == 422:
time.sleep(0.3)
continue
r.raise_for_status()
raise RuntimeError("Failed to create vmail.dev mailbox after 10 attempts")
def wait_for_otp(self, mailbox: dict, timeout: int = 120, poll: float = 5.0) -> str:
"""Poll mailbox until a 6-digit OTP arrives."""
mailbox_id = mailbox["id"]
if not mailbox_id or mailbox_id == "existing":
# Try to find the mailbox ID by creating a new one with same local part
mailbox_id = self._find_or_create_mailbox(mailbox["address"])
if not mailbox_id:
raise TimeoutError(f"Could not find mailbox ID for {mailbox['address']}")
deadline = time.time() + timeout
while time.time() < deadline:
try:
r = httpx.get(
f"{self.base}/mailboxes/{mailbox_id}/messages",
headers=self.headers,
timeout=30,
)
r.raise_for_status()
messages = r.json().get("data", [])
except httpx.HTTPError as exc:
print(f"mail poll error: {exc.__class__.__name__}")
time.sleep(poll)
continue
if messages:
try:
msg_r = httpx.get(
f"{self.base}/mailboxes/{mailbox_id}/messages/{messages[0]['id']}",
headers=self.headers,
timeout=30,
)
msg_r.raise_for_status()
msg = msg_r.json()["data"]
except httpx.HTTPError as exc:
print(f"mail fetch error: {exc.__class__.__name__}")
time.sleep(poll)
continue
body = msg.get("text") or msg.get("html") or ""
if isinstance(body, list):
body = "\n".join(body)
codes = re.findall(r"\b(\d{6})\b", body)
if codes:
return codes[0]
time.sleep(poll)
raise TimeoutError(f"No OTP received within {timeout}s")
def _find_or_create_mailbox(self, address: str) -> str | None:
"""Try to find existing mailbox or create new one for OTP fetching."""
local_part = address.split("@")[0]
# Try creating with the exact local part - if it exists, we get 409
r = httpx.post(
f"{self.base}/mailboxes",
headers=self.headers,
json={"localPart": local_part},
timeout=30,
)
if r.status_code == 201:
return r.json()["data"]["id"]
if r.status_code == 409:
# Mailbox exists with this API key but we don't know the ID
# Create a new temporary mailbox and use that for OTP
temp_local = f"{local_part}-{secrets.token_hex(4)}"
r2 = httpx.post(
f"{self.base}/mailboxes",
headers=self.headers,
json={"localPart": temp_local},
timeout=30,
)
if r2.status_code == 201:
return r2.json()["data"]["id"]
return None
class MailTMClient(BaseMailClient):
"""mail.tm temporary email client."""
def __init__(self):
self.base = MAILTM_BASE
self._token = None
def create_mailbox(self) -> dict:
"""Create a mailbox. Returns dict with id, address, password, token."""
domains_resp = httpx.get(f"{self.base}/domains", timeout=30)
domains_resp.raise_for_status()
domains = domains_resp.json().get("hydra:member", [])
active = next((d for d in domains if d.get("isActive")), None)
@@ -23,13 +162,13 @@ class MailClient:
address = f"{secrets.token_hex(6)}@{active['domain']}"
password = secrets.token_urlsafe(18)
r = httpx.post(
f"{MAILTM_BASE}/accounts",
f"{self.base}/accounts",
json={"address": address, "password": password},
timeout=30,
)
if r.status_code == 201:
token_resp = httpx.post(
f"{MAILTM_BASE}/token",
f"{self.base}/token",
json={"address": address, "password": password},
timeout=30,
)
@@ -47,14 +186,29 @@ class MailClient:
raise RuntimeError("Failed to create mail.tm mailbox after 10 attempts")
def get_token_for_address(self, address: str, password: str) -> str:
"""Get auth token for an existing mailbox (for mail.tm only)."""
r = httpx.post(
f"{self.base}/token",
json={"address": address, "password": password},
timeout=30,
)
if r.status_code == 200:
return r.json()["token"]
return ""
def wait_for_otp(self, mailbox: dict, timeout: int = 120, poll: float = 5.0) -> str:
"""Poll mailbox until a 6-digit OTP arrives."""
headers = {"Authorization": f"Bearer {mailbox['token']}"}
token = mailbox.get("token", "")
if not token:
raise ValueError("mail.tm requires token for wait_for_otp")
headers = {"Authorization": f"Bearer {token}"}
deadline = time.time() + timeout
while time.time() < deadline:
try:
r = httpx.get(
f"{MAILTM_BASE}/messages",
f"{self.base}/messages",
headers=headers,
timeout=30,
)
@@ -68,7 +222,7 @@ class MailClient:
if messages:
try:
msg_r = httpx.get(
f"{MAILTM_BASE}/messages/{messages[0]['id']}",
f"{self.base}/messages/{messages[0]['id']}",
headers=headers,
timeout=30,
)
@@ -89,3 +243,117 @@ class MailClient:
time.sleep(poll)
raise TimeoutError(f"No OTP received within {timeout}s")
class YYDSMailClient(BaseMailClient):
"""YYDS Mail temporary email client (maliapi.215.im)."""
def __init__(self):
self.api_key = YYDS_API_KEY
self.headers = {"X-API-Key": self.api_key}
self.base = YYDS_BASE
def create_mailbox(self) -> dict:
"""Create a mailbox. Returns dict with id, address, token."""
for attempt in range(10):
r = httpx.post(
f"{self.base}/v1/accounts",
headers=self.headers,
json={},
timeout=30,
)
if r.status_code == 201 or r.status_code == 200:
data = r.json()["data"]
return {
"id": data["id"],
"address": data["address"],
"password": "",
"token": data["token"],
}
if r.status_code == 429:
time.sleep(1)
continue
r.raise_for_status()
raise RuntimeError("Failed to create YYDS mailbox after 10 attempts")
def get_token_for_address(self, address: str, password: str = "") -> str:
"""Get auth token for an existing mailbox via POST /v1/token."""
r = httpx.post(
f"{self.base}/v1/token",
headers=self.headers,
json={"address": address},
timeout=30,
)
if r.status_code == 200:
return r.json()["data"]["token"]
return ""
def wait_for_otp(self, mailbox: dict, timeout: int = 120, poll: float = 5.0) -> str:
"""Poll mailbox until a 6-digit OTP arrives."""
address = mailbox.get("address", "")
token = mailbox.get("token", "")
# If no token, try to get one from address
if not token and address:
token = self.get_token_for_address(address)
if not token:
raise TimeoutError(f"Could not get YYDS token for {address}")
headers = {"Authorization": f"Bearer {token}"}
deadline = time.time() + timeout
while time.time() < deadline:
try:
r = httpx.get(
f"{self.base}/v1/messages",
headers=headers,
params={"address": address},
timeout=30,
)
r.raise_for_status()
result = r.json().get("data", {})
messages = result.get("messages", [])
except httpx.HTTPError as exc:
print(f"mail poll error: {exc.__class__.__name__}")
time.sleep(poll)
continue
if messages:
try:
msg_r = httpx.get(
f"{self.base}/v1/messages/{messages[0]['id']}",
headers=headers,
timeout=30,
)
msg_r.raise_for_status()
msg = msg_r.json().get("data", {})
except httpx.HTTPError as exc:
print(f"mail fetch error: {exc.__class__.__name__}")
time.sleep(poll)
continue
text = msg.get("text", "") or ""
html_list = msg.get("html", [])
if isinstance(html_list, list):
html = "\n".join(html_list)
else:
html = html_list
body = text or html or ""
codes = re.findall(r"\b(\d{6})\b", body)
if codes:
return codes[0]
time.sleep(poll)
raise TimeoutError(f"No OTP received within {timeout}s")
def get_mail_client(provider: str = "vmail") -> BaseMailClient:
"""Get mail client by provider name."""
if provider == "mailtm":
return MailTMClient()
if provider == "yyds":
return YYDSMailClient()
return VMailClient() # default to vmail
# Default instance
MailClient = get_mail_client("vmail")