feat(llm): Add subscription-based authentication for OpenAI Codex models

examples/01_standalone_sdk/35_subscription_login.py

@@ -0,0 +1,61 @@
+"""Example: Using ChatGPT subscription for Codex models.
+
+This example demonstrates how to use your ChatGPT Plus/Pro subscription
+to access OpenAI's Codex models without consuming API credits.
+
+The subscription_login() method handles:
+- OAuth PKCE authentication flow
+- Credential caching (~/.openhands/auth/)
+- Automatic token refresh
+
+Supported models:
+- gpt-5.2-codex
+- gpt-5.2
+- gpt-5.1-codex-max
+- gpt-5.1-codex-mini
+
+Requirements:
+- Active ChatGPT Plus or Pro subscription
+- Browser access for initial OAuth login
+"""
+
+import os
+
+from openhands.sdk import LLM, Agent, Conversation, Tool
+from openhands.tools.file_editor import FileEditorTool
+from openhands.tools.terminal import TerminalTool
+
+
+# First time: Opens browser for OAuth login
+# Subsequent calls: Reuses cached credentials (auto-refreshes if expired)
+llm = LLM.subscription_login(
+    vendor="openai",
+    model="gpt-5.2-codex",  # or "gpt-5.2", "gpt-5.1-codex-max", "gpt-5.1-codex-mini"
+)
+
+# Alternative: Force a fresh login (useful if credentials are stale)
+# llm = LLM.subscription_login(vendor="openai", model="gpt-5.2-codex", force_login=True)
+
+# Alternative: Disable auto-opening browser (prints URL to console instead)
+# llm = LLM.subscription_login(
+#     vendor="openai", model="gpt-5.2-codex", open_browser=False
+# )
+
+# Verify subscription mode is active
+print(f"Using subscription mode: {llm.is_subscription}")
+
+# Use the LLM with an agent as usual
+agent = Agent(
+    llm=llm,
+    tools=[
+        Tool(name=TerminalTool.name),
+        Tool(name=FileEditorTool.name),
+    ],
+)
+
+cwd = os.getcwd()
+conversation = Conversation(agent=agent, workspace=cwd)
+
+conversation.send_message("List the files in the current directory.")
+conversation.run()
+print("Done!")

openhands-sdk/openhands/sdk/llm/__init__.py

@@ -1,3 +1,9 @@
+from openhands.sdk.llm.auth import (
+    OPENAI_CODEX_MODELS,
+    CredentialStore,
+    OAuthCredentials,
+    OpenAISubscriptionAuth,
+)
 from openhands.sdk.llm.llm import LLM
 from openhands.sdk.llm.llm_registry import LLMRegistry, RegistryEvent
 from openhands.sdk.llm.llm_response import LLMResponse
@@ -22,11 +28,18 @@
 
 
 __all__ = [
+    # Auth
+    "CredentialStore",
+    "OAuthCredentials",
+    "OpenAISubscriptionAuth",
+    "OPENAI_CODEX_MODELS",
+    # Core
     "LLMResponse",
     "LLM",
     "LLMRegistry",
     "RouterLLM",
     "RegistryEvent",
+    # Messages
     "Message",
     "MessageToolCall",
     "TextContent",
@@ -35,10 +48,13 @@
     "RedactedThinkingBlock",
     "ReasoningItemModel",
     "content_to_str",
+    # Streaming
     "LLMStreamChunk",
     "TokenCallbackType",
+    # Metrics
     "Metrics",
     "MetricsSnapshot",
+    # Models
     "VERIFIED_MODELS",
     "UNVERIFIED_MODELS_EXCLUDING_BEDROCK",
     "get_unverified_models",

openhands-sdk/openhands/sdk/llm/auth/__init__.py

@@ -0,0 +1,28 @@
+"""Authentication module for LLM subscription-based access.
+
+This module provides OAuth-based authentication for LLM providers that support
+subscription-based access (e.g., ChatGPT Plus/Pro for OpenAI Codex models).
+"""
+
+from openhands.sdk.llm.auth.credentials import (
+    CredentialStore,
+    OAuthCredentials,
+)
+from openhands.sdk.llm.auth.openai import (
+    OPENAI_CODEX_MODELS,
+    OpenAISubscriptionAuth,
+    SupportedVendor,
+    inject_system_prefix,
+    transform_for_subscription,
+)
+
+
+__all__ = [
+    "CredentialStore",
+    "OAuthCredentials",
+    "OpenAISubscriptionAuth",
+    "OPENAI_CODEX_MODELS",
+    "SupportedVendor",
+    "inject_system_prefix",
+    "transform_for_subscription",
+]

openhands-sdk/openhands/sdk/llm/auth/credentials.py

@@ -0,0 +1,157 @@
+"""Credential storage and retrieval for OAuth-based LLM authentication."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import warnings
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from openhands.sdk.logger import get_logger
+
+
+logger = get_logger(__name__)
+
+
+def get_credentials_dir() -> Path:
+    """Get the directory for storing credentials.
+
+    Uses XDG_DATA_HOME if set, otherwise defaults to ~/.local/share/openhands.
+    """
+    return Path.home() / ".openhands" / "auth"
+
+
+class OAuthCredentials(BaseModel):
+    """OAuth credentials for subscription-based LLM access."""
+
+    type: Literal["oauth"] = "oauth"
+    vendor: str = Field(description="The vendor/provider (e.g., 'openai')")
+    access_token: str = Field(description="The OAuth access token")
+    refresh_token: str = Field(description="The OAuth refresh token")
+    expires_at: int = Field(
+        description="Unix timestamp (ms) when the access token expires"
+    )
+
+    def is_expired(self) -> bool:
+        """Check if the access token is expired."""
+        # Add 60 second buffer to avoid edge cases
+        # Add 60 second buffer to avoid edge cases where token expires during request
+        return self.expires_at < (int(time.time() * 1000) + 60_000)
+
+
+class CredentialStore:
+    """Store and retrieve OAuth credentials for LLM providers."""
+
+    def __init__(self, credentials_dir: Path | None = None):
+        """Initialize the credential store.
+
+        Args:
+            credentials_dir: Optional custom directory for storing credentials.
+                           Defaults to ~/.local/share/openhands/auth/
+        """
+        self._credentials_dir = credentials_dir or get_credentials_dir()
+        logger.info(f"Using credentials directory: {self._credentials_dir}")
+
+    @property
+    def credentials_dir(self) -> Path:
+        """Get the credentials directory, creating it if necessary."""
+        self._credentials_dir.mkdir(parents=True, exist_ok=True)
+        # Set directory permissions to owner-only (rwx------)
+        if os.name != "nt":
+            self._credentials_dir.chmod(0o700)
+        return self._credentials_dir
+
+    def _get_credentials_file(self, vendor: str) -> Path:
+        """Get the path to the credentials file for a vendor."""
+        return self.credentials_dir / f"{vendor}_oauth.json"
+
+    def get(self, vendor: str) -> OAuthCredentials | None:
+        """Get stored credentials for a vendor.
+
+        Args:
+            vendor: The vendor/provider name (e.g., 'openai')
+
+        Returns:
+            OAuthCredentials if found and valid, None otherwise
+        """
+        creds_file = self._get_credentials_file(vendor)
+        if not creds_file.exists():
+            return None
+
+        try:
+            with open(creds_file) as f:
+                data = json.load(f)
+            return OAuthCredentials.model_validate(data)
+        except (json.JSONDecodeError, ValueError):
+            # Invalid credentials file, remove it
+            creds_file.unlink(missing_ok=True)
+            return None
+
+    def save(self, credentials: OAuthCredentials) -> None:
+        """Save credentials for a vendor.
+
+        Args:
+            credentials: The OAuth credentials to save
+        """
+        creds_file = self._get_credentials_file(credentials.vendor)
+        with open(creds_file, "w") as f:
+            json.dump(credentials.model_dump(), f, indent=2)
+        # Set restrictive permissions (owner read/write only)
+        # Note: On Windows, NTFS ACLs should be used instead
+        if os.name != "nt":  # Not Windows
+            creds_file.chmod(0o600)
+        else:
+            warnings.warn(
+                "File permissions on Windows should be manually restricted",
+                stacklevel=2,
+            )
+
+    def delete(self, vendor: str) -> bool:
+        """Delete stored credentials for a vendor.
+
+        Args:
+            vendor: The vendor/provider name
+
+        Returns:
+            True if credentials were deleted, False if they didn't exist
+        """
+        creds_file = self._get_credentials_file(vendor)
+        if creds_file.exists():
+            creds_file.unlink()
+            return True
+        return False
+
+    def update_tokens(
+        self,
+        vendor: str,
+        access_token: str,
+        refresh_token: str | None,
+        expires_in: int,
+    ) -> OAuthCredentials | None:
+        """Update tokens for an existing credential.
+
+        Args:
+            vendor: The vendor/provider name
+            access_token: New access token
+            refresh_token: New refresh token (if provided)
+            expires_in: Token expiry in seconds
+
+        Returns:
+            Updated credentials, or None if no existing credentials found
+        """
+        existing = self.get(vendor)
+        if existing is None:
+            return None
+
+        updated = OAuthCredentials(
+            vendor=vendor,
+            access_token=access_token,
+            refresh_token=refresh_token or existing.refresh_token,
+            expires_at=int(time.time() * 1000) + (expires_in * 1000),
+        )
+        self.save(updated)
+        return updated