codex-console-automation

# codex-console Automation Skill > Skill by [ara.so](https://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 1. Start the web UI: ```bash python webui.py --access-password mypassword ``` 2. Access at `http://127.0.0.1:8000` 3. 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:** 1. Sentinel POW solving 2. Email verification (auto-fetch from email service) 3. Split registration/login flow 4. Token extraction 5. 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)