codex-console Automation Skill

Skill by ara.so — Codex Skills collection.

Overview

codex-console is a comprehensive automation platform for managing OpenAI accounts at scale. It handles registration, login, token extraction, subscription management, payment binding, auto-replenishment, and integration with API gateways like CPA, Sub2API, Team Manager, and New-API.

This is a maintained fork that fixes compatibility issues with OpenAI's evolving authentication flow, including Sentinel POW solving, split registration/login flows, and improved OTP handling.

Key capabilities:

Web UI for task management and monitoring
Batch account registration with email service integration
Semi-automated payment card binding with 3DS support
Automated token refresh and upload to API gateways
Self-check and repair system
Auto-replenishment based on inventory levels
Tag-based account pooling and team management
SQLite or PostgreSQL backend

Installation

Using Python (Recommended for Development)

bash

# Clone the repository
git clone https://github.com/dou-jiang/codex-console.git
cd codex-console

# Install dependencies with uv (recommended)
uv sync

# Or use pip
pip install -r requirements.txt

Using Docker

bash

# With docker-compose
docker-compose up -d

# Or with docker run
docker run -d \
  -p 1455:1455 \
  -p 6080:6080 \
  -e WEBUI_HOST=0.0.0.0 \
  -e WEBUI_PORT=1455 \
  -e WEBUI_ACCESS_PASSWORD=your_secure_password \
  -v $(pwd)/data:/app/data \
  --name codex-console \
  ghcr.io/dou-jiang/codex-console:latest

Building Standalone Executable

bash

# Windows
build.bat

# Linux/macOS
bash build.sh

The executable will be in

dist/codex-console-windows-X64.exe

or equivalent.

Configuration

Environment Variables

Create

.env

from template:

bash

cp .env.example .env

Key variables:

bash

# Server
APP_HOST=0.0.0.0
APP_PORT=8000
APP_ACCESS_PASSWORD=admin123

# Database (SQLite by default)
APP_DATABASE_URL=data/database.db

# Or PostgreSQL
APP_DATABASE_URL=postgresql://user:password@host:5432/dbname

# Logging
LOG_LEVEL=info

# Debug mode
DEBUG=false

Priority: CLI args >

.env

> database settings > defaults

First-Time Setup

Start the web UI:

bash

python webui.py --access-password mypassword

Access at
```
http://127.0.0.1:8000
```
Configure in Settings page:
- Email service credentials
- Proxy settings
- Upload targets (CPA/Sub2API/New-API)
- Payment automation options

Core Usage Patterns

Starting the Web UI

python

# webui.py - main entry point
import uvicorn
from src.web.app import app
from src.utils.settings import get_settings

if __name__ == "__main__":
    settings = get_settings()
    uvicorn.run(
        app,
        host=settings.app_host,
        port=settings.app_port,
        log_level=settings.log_level.lower()
    )

CLI options:

bash

# Default
python webui.py

# Custom host/port
python webui.py --host 0.0.0.0 --port 8080

# Set password
python webui.py --access-password mypassword

# Debug mode (hot reload)
python webui.py --debug

Email Service Integration

codex-console supports multiple email providers:

python

# src/services/email_service.py
from src.services.email_service import EmailServiceFactory

# Configure in database or code
email_config = {
    "service_type": "cloudmail",  # or luckmail, yydsmail, outlook
    "api_key": None,  # from env: CLOUDMAIL_API_KEY
    "config": {
        "base_url": "https://api.cloudmail.com",
        "timeout": 30
    }
}

# Factory creates appropriate service
service = EmailServiceFactory.create(email_config)

# Fetch verification code
otp = await service.get_verification_code(
    email="test@example.com",
    timeout=60
)

Supported services:

CloudMail (API-based)
LuckMail (API-based)
YYDS Mail (API-based)
Outlook (self-hosted accounts)

Registration Flow

python

# src/core/register.py
from src.core.register import RegisterService
from src.models.task import RegisterTask

async def register_account(email: str, password: str, proxy: str):
    """Register a new OpenAI account"""
    
    task = RegisterTask(
        email=email,
        password=password,
        proxy=proxy,
        status="pending"
    )
    
    service = RegisterService(task)
    result = await service.execute()
    
    return result
    # Returns: {"success": True, "account_id": 123, "token": "..."}

Key steps handled:

Sentinel POW solving
Email verification (auto-fetch from email service)
Split registration/login flow
Token extraction
Workspace caching

Batch Registration

python

# src/core/auto_register.py
from src.core.auto_register import AutoRegisterService

async def batch_register(count: int):
    """Register multiple accounts"""
    
    service = AutoRegisterService()
    
    # Configure
    await service.configure({
        "target_count": count,
        "email_service": "cloudmail",
        "proxy_pool": "residential",
        "auto_upload": True,
        "upload_target": "newapi"
    })
    
    # Start registration
    await service.start()
    
    # Monitor progress
    status = await service.get_status()
    # Returns: {"completed": 50, "failed": 2, "running": True}

Payment Binding

python

# src/core/payment.py
from src.core.payment import PaymentService
from src.models.task import BindCardTask

async def bind_payment_card(account_id: int, card_info: dict):
    """Bind payment card to account (semi-automated)"""
    
    task = BindCardTask(
        account_id=account_id,
        card_number=card_info["number"],  # Store encrypted
        expiry=card_info["expiry"],
        cvv=card_info["cvv"],
        billing_address={
            "street": "auto-generated",  # Random US address
            "city": "New York",
            "state": "NY",
            "zip": "10001",
            "country": "US"
        },
        status="pending"
    )
    
    service = PaymentService(task)
    result = await service.execute()
    
    # Note: 3DS verification requires manual intervention
    # Browser window will open for 3DS flow
    
    return result

Token Management

python

# src/core/account_manager.py
from src.core.account_manager import AccountManager
from src.models.account import Account

async def refresh_account_token(account_id: int):
    """Refresh access token for an account"""
    
    manager = AccountManager()
    account = await manager.get_account(account_id)
    
    # Refresh using refresh_token
    new_token = await manager.refresh_token(account)
    
    # Auto-upload to configured targets
    await manager.upload_to_targets(account)
    
    return new_token

Auto-Upload to API Gateways

python

# src/services/upload_service.py
from src.services.upload_service import UploadService

async def upload_account(account_id: int, target: str):
    """Upload account to API gateway"""
    
    service = UploadService()
    
    # Supported targets: cpa, sub2api, team_manager, newapi
    result = await service.upload(
        account_id=account_id,
        target=target
    )
    
    return result
    # Returns: {"success": True, "external_id": "...", "quota": 20}

New-API configuration example:

python

# In settings or database
newapi_config = {
    "base_url": "https://api.example.com",
    "api_key": None,  # from env: NEWAPI_API_KEY
    "auto_upload": True,
    "upload_on_register": True,
    "quota_per_account": 20
}

Task Management

python

# src/services/task_service.py
from src.services.task_service import TaskService

async def manage_tasks():
    """Unified task management"""
    
    service = TaskService()
    
    # Create registration task
    task_id = await service.create_task(
        task_type="register",
        count=10,
        config={"email_service": "cloudmail"}
    )
    
    # Pause task
    await service.pause_task(task_id)
    
    # Resume task
    await service.resume_task(task_id)
    
    # Cancel task
    await service.cancel_task(task_id)
    
    # Retry failed items
    await service.retry_task(task_id)
    
    # Get status
    status = await service.get_task_status(task_id)

Auto-Replenishment

python

# src/core/auto_replenishment.py
from src.core.auto_replenishment import AutoReplenishmentService

async def configure_auto_replenish():
    """Configure automatic account replenishment"""
    
    service = AutoReplenishmentService()
    
    await service.configure({
        "enabled": True,
        "min_threshold": 10,  # Start when inventory < 10
        "target_count": 50,   # Replenish to 50 accounts
        "check_interval": 3600,  # Check every hour
        "max_daily_registers": 100
    })
    
    # Start monitoring
    await service.start()

Self-Check and Repair

python

# src/services/selfcheck.py
from src.services.selfcheck import SelfCheckService

async def run_system_selfcheck():
    """Run comprehensive system self-check"""
    
    service = SelfCheckService()
    
    # Run checks
    results = await service.run_all_checks()
    
    # Results include:
    # - Database connectivity
    # - Email service status
    # - Proxy availability
    # - Account token validity
    # - Upload target connectivity
    
    # Auto-repair if enabled
    if results["has_issues"]:
        await service.auto_repair(results)
    
    return results

Account Pooling and Tags

python

# src/models/account.py
from src.core.account_manager import AccountManager

async def manage_account_pools():
    """Organize accounts with tags and pools"""
    
    manager = AccountManager()
    
    # Add account to team pool
    await manager.update_account(
        account_id=123,
        updates={
            "role_tag": "team_member",
            "biz_tag": "project_alpha",
            "pool_state": "team_pool",
            "priority": 5
        }
    )
    
    # Query by pool
    team_accounts = await manager.get_accounts_by_pool("team_pool")
    
    # Query by tag
    project_accounts = await manager.get_accounts_by_tag("project_alpha")

Database Models

Account Model

python

# src/models/account.py
from sqlalchemy import Column, Integer, String, DateTime, JSON
from src.database import Base

class Account(Base):
    __tablename__ = "accounts"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    password = Column(String, nullable=False)
    
    # Auth tokens
    access_token = Column(String)
    refresh_token = Column(String)
    session_token = Column(String)
    
    # Status
    status = Column(String, default="active")  # active, suspended, expired
    subscription_status = Column(String)  # none, trial, plus, team
    
    # Pooling and tags
    role_tag = Column(String)  # team_member, candidate, etc.
    biz_tag = Column(String)   # project/client identifier
    pool_state = Column(String)  # team_pool, candidate_pool, blocked_pool
    priority = Column(Integer, default=0)
    
    # Metadata
    quota = Column(Integer, default=0)
    workspace_id = Column(String)
    last_used_at = Column(DateTime)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)
    
    # Upload tracking
    upload_status = Column(JSON)  # {"cpa": true, "newapi": true}

Task Model

python

# src/models/task.py
from sqlalchemy import Column, Integer, String, DateTime, JSON
from src.database import Base

class RegisterTask(Base):
    __tablename__ = "register_tasks"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, nullable=False)
    password = Column(String, nullable=False)
    proxy = Column(String)
    
    # Task state
    status = Column(String, default="pending")  # pending, running, completed, failed, paused
    progress = Column(Integer, default=0)
    error_message = Column(String)
    
    # Result
    account_id = Column(Integer)  # Reference to created account
    access_token = Column(String)
    
    # Metadata
    created_at = Column(DateTime)
    started_at = Column(DateTime)
    completed_at = Column(DateTime)
    retry_count = Column(Integer, default=0)

API Routes

Account Management

python

# GET /api/accounts
# List all accounts with filters
GET /api/accounts?status=active&pool=team_pool&limit=50

# GET /api/accounts/{id}
# Get account details
GET /api/accounts/123

# POST /api/accounts/{id}/refresh
# Refresh account token
POST /api/accounts/123/refresh

# POST /api/accounts/{id}/upload
# Upload to API gateway
POST /api/accounts/123/upload
{
    "target": "newapi"
}

# DELETE /api/accounts/{id}
# Delete account
DELETE /api/accounts/123

Task Management

python

# POST /api/tasks/register
# Create registration task
POST /api/tasks/register
{
    "count": 10,
    "email_service": "cloudmail",
    "auto_upload": true,
    "upload_target": "newapi"
}

# POST /api/tasks/{id}/pause
# Pause task
POST /api/tasks/123/pause

# POST /api/tasks/{id}/resume
# Resume task
POST /api/tasks/123/resume

# POST /api/tasks/{id}/cancel
# Cancel task
POST /api/tasks/123/cancel

# POST /api/tasks/{id}/retry
# Retry failed items
POST /api/tasks/123/retry

Export and Import

python

# GET /api/export/accounts
# Export accounts in various formats
GET /api/export/accounts?format=codex&pool=team_pool

# Supported formats: codex, json, csv, newapi, cpa

# POST /api/import/accounts
# Import accounts from file
POST /api/import/accounts
Content-Type: multipart/form-data
file: accounts.json

Common Workflows

Complete Registration Pipeline

python

import asyncio
from src.core.register import RegisterService
from src.services.upload_service import UploadService
from src.models.task import RegisterTask

async def full_registration_workflow():
    """End-to-end: register account, verify, bind card, upload"""
    
    # Step 1: Register
    task = RegisterTask(
        email="auto-generated@cloudmail.com",
        password="SecurePass123!",
        proxy="http://proxy.example.com:8080"
    )
    
    reg_service = RegisterService(task)
    result = await reg_service.execute()
    
    if not result["success"]:
        raise Exception(f"Registration failed: {result['error']}")
    
    account_id = result["account_id"]
    
    # Step 2: Bind payment (semi-automated)
    # Note: This opens browser for 3DS - requires monitoring
    from src.core.payment import PaymentService
    from src.models.task import BindCardTask
    
    bind_task = BindCardTask(
        account_id=account_id,
        card_number="encrypted_card_data",
        expiry="12/25",
        cvv="123",
        billing_address={"auto": True}  # Auto-generate random US address
    )
    
    payment_service = PaymentService(bind_task)
    payment_result = await payment_service.execute()
    
    # Step 3: Upload to API gateway
    upload_service = UploadService()
    upload_result = await upload_service.upload(
        account_id=account_id,
        target="newapi"
    )
    
    return {
        "account_id": account_id,
        "email": task.email,
        "uploaded": upload_result["success"]
    }

# Run
asyncio.run(full_registration_workflow())

Batch Account Refresh

python

from src.core.account_manager import AccountManager
import asyncio

async def batch_refresh_tokens():
    """Refresh tokens for all active accounts"""
    
    manager = AccountManager()
    
    # Get all active accounts
    accounts = await manager.get_accounts(status="active")
    
    results = []
    for account in accounts:
        try:
            new_token = await manager.refresh_token(account)
            
            # Auto-upload if configured
            if manager.settings.auto_upload_on_refresh:
                await manager.upload_to_targets(account)
            
            results.append({
                "account_id": account.id,
                "success": True,
                "token": new_token
            })
        except Exception as e:
            results.append({
                "account_id": account.id,
                "success": False,
                "error": str(e)
            })
    
    return results

# Run
asyncio.run(batch_refresh_tokens())

Scheduled Auto-Replenishment

python

# src/schedulers/auto_replenish_scheduler.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from src.core.auto_replenishment import AutoReplenishmentService

scheduler = AsyncIOScheduler()

async def check_and_replenish():
    """Check inventory and trigger replenishment if needed"""
    
    service = AutoReplenishmentService()
    status = await service.check_inventory()
    
    if status["current_count"] < status["min_threshold"]:
        needed = status["target_count"] - status["current_count"]
        await service.start_replenishment(count=needed)

# Schedule every hour
scheduler.add_job(
    check_and_replenish,
    'interval',
    hours=1,
    id='auto_replenish'
)

scheduler.start()

Troubleshooting

Sentinel POW Solving Fails

Issue: Registration fails with "Sentinel POW required" error.

Solution: Ensure you're using latest version (v1.1.2+) which includes POW solving:

python

# src/core/register.py automatically handles this
# If issues persist, check proxy quality:
from src.utils.proxy import test_proxy

result = await test_proxy("http://proxy.example.com:8080")
if not result["sentinel_pass"]:
    print("Proxy blocked by Sentinel - use residential proxy")

Email OTP Not Received

Issue: Registration hangs waiting for verification code.

Solution:

Check email service status in Settings
Verify API key is set correctly
Test email service manually:

python

from src.services.email_service import EmailServiceFactory

service = EmailServiceFactory.create({
    "service_type": "cloudmail",
    "api_key": "test_key"
})

# Test connection
status = await service.test_connection()
print(status)  # Should return {"success": True}

Token Refresh Fails

Issue:

refresh_token

returns 401 Unauthorized.

Solution:

Check if refresh token is expired (30 days typically)
Force re-login instead:

python

from src.core.account_manager import AccountManager

manager = AccountManager()
account = await manager.get_account(account_id)

# Re-login to get fresh tokens
await manager.re_login(account)

Payment Binding Stuck on 3DS

Issue: Browser opens for 3DS but verification never completes.

Solution:

3DS cannot be automated. Monitor the browser window and:

Complete 3DS challenge manually
Wait for callback
codex-console will detect completion and continue

Database Migration Issues

Issue: Alembic migration fails after update.

Solution:

bash

# Reset to latest migration
alembic stamp head

# Or manually migrate
alembic upgrade head

# If corrupted, backup and recreate:
cp data/database.db data/database.db.backup
rm data/database.db
python webui.py  # Auto-creates with latest schema

Auto-Upload Not Working

Issue: Accounts registered but not uploaded to New-API/CPA.

Solution:

Verify upload target is configured in Settings
Test connection:

python

from src.services.upload_service import UploadService

service = UploadService()
test_result = await service.test_target("newapi")
print(test_result)  # Check connectivity

Enable auto-upload in task config:

python

{
    "auto_upload": true,
    "upload_target": "newapi",
    "upload_on_register": true
}

High Registration Failure Rate

Issue: Most registration attempts fail.

Solution:

Use residential proxies - datacenter IPs are often blocked
Slow down - add delays between attempts
Check email service quota - may be rate-limited
Review logs - look for specific error patterns

bash

# Enable debug logging
export LOG_LEVEL=debug
python webui.py

# Check logs in data/logs/
tail -f data/logs/register.log

Port Already in Use

Issue:

Address already in use

error on startup.

Solution:

codex-console auto-switches ports if 8000 is taken. To force a specific port:

bash

python webui.py --port 8080

Or kill the process using port 8000:

bash

# Linux/Mac
lsof -ti:8000 | xargs kill -9

# Windows
netstat -ano | findstr :8000
taskkill /PID <PID> /F

Advanced Configuration

Custom Email Service

Implement custom email provider:

python

# src/services/custom_email.py
from src.services.email_service import BaseEmailService

class CustomEmailService(BaseEmailService):
    async def get_verification_code(self, email: str, timeout: int = 60) -> str:
        """Fetch OTP from custom provider"""
        # Your implementation
        pass
    
    async def test_connection(self) -> dict:
        """Test API connectivity"""
        # Your implementation
        pass

# Register in factory
# src/services/email_service.py
EMAIL_SERVICES["custom"] = CustomEmailService

Custom Upload Target

Add new API gateway integration:

python

# src/services/custom_upload.py
from src.services.upload_service import BaseUploadTarget

class CustomUploadTarget(BaseUploadTarget):
    async def upload_account(self, account: Account) -> dict:
        """Upload account to custom gateway"""
        # Your implementation
        pass

# Register
# src/services/upload_service.py
UPLOAD_TARGETS["custom"] = CustomUploadTarget

Security Best Practices

Never commit secrets - use environment variables:

bash

export CLOUDMAIL_API_KEY=your_key_here
export NEWAPI_API_KEY=your_key_here

Change default password immediately:

bash

python webui.py --access-password strong_password_here

Use PostgreSQL for production - SQLite is development-only:

bash

export APP_DATABASE_URL=postgresql://user:pass@host:5432/db

Encrypt sensitive fields - card data, passwords are encrypted at rest
Enable audit logging - track all operations:

python

from src.models.audit import OperationAuditLog

# Auto-logged for sensitive operations

Resources

Official Repo: https://github.com/dou-jiang/codex-console
Blog/Docs: https://blog.cysq8.cn/
QQ Group: 291638849
Telegram: https://t.me/codex_console
License: MIT

codex-console-automation

NPX Install

Tags

SKILL.md Content

codex-console Automation Skill

Overview

Installation

Using Python (Recommended for Development)

Using Docker

Building Standalone Executable

Configuration

Environment Variables

First-Time Setup

Core Usage Patterns

Starting the Web UI

Email Service Integration

Registration Flow

Batch Registration

Payment Binding

Token Management

Auto-Upload to API Gateways

Task Management

Auto-Replenishment

Self-Check and Repair

Account Pooling and Tags

Database Models

Account Model

Task Model

API Routes

Account Management

Task Management

Export and Import

Common Workflows

Complete Registration Pipeline

Batch Account Refresh

Scheduled Auto-Replenishment

Troubleshooting

Sentinel POW Solving Fails

Email OTP Not Received

Token Refresh Fails

Payment Binding Stuck on 3DS

Database Migration Issues

Auto-Upload Not Working

High Registration Failure Rate

Port Already in Use

Advanced Configuration

Custom Email Service

Custom Upload Target

Security Best Practices

Resources