oauth.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/openai_compat/oauth.py
 
"""OpenRouter paste-only PKCE authentication.
 
Purpose
-------
Exchange a manually pasted authorization code for a provider key.
 
Responsibilities
----------------
Generate PKCE, display the URL through callbacks, and enforce transport policy.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 - lifecycle
    L3 - semantic details without credentials
    L4 - module lifecycle
 
Contracts
---------
No listener, browser launch, refresh, proxy inheritance or redirect following.
The operator approved this fixed-endpoint sink outside ClientInterface for P6.
Credentials and transport exception contents never reach output channels.
 
Dependencies
------------
- oc_diagnostics (diagnostics)
- httpx (existing provider transport dependency)
- option_c_agent_contracts.provider_plugin (OAuth protocols and credentials)
"""
from __future__ import annotations
 
import base64
import hashlib
import secrets
import time
from types import MappingProxyType
from typing import Final, Mapping
from urllib.parse import parse_qs, urlencode, urlsplit
 
import httpx
from oc_diagnostics import _dbg
from option_c_agent_contracts.provider_plugin import OAuthCredentials, OAuthLoginCallbacks
 
_dbg('OC_CHAT.CLIENT', 'module', 'loaded', 4)
 
TOKEN_ENDPOINTS: Final[Mapping[str, str]] = MappingProxyType({
    'openrouter': 'https://openrouter.ai/api/v1/auth/keys',
})
CALLBACK_URL: Final = 'http://localhost:3000/'
AUTHORIZE_URL: Final = 'https://openrouter.ai/auth'
 
 
def _pasted_code(value: str) -> str:
    """Accept a bare code or the exact callback URL copied from the browser."""
    text = value.strip()
    if '://' in text:
        parts = urlsplit(text)
        callback = urlsplit(CALLBACK_URL)
        if (parts.scheme, parts.netloc, parts.path) != (callback.scheme, callback.netloc, callback.path):
            raise ValueError('Unexpected OAuth callback URL.')
        codes = parse_qs(parts.query).get('code', [])
        if len(codes) != 1:
            raise ValueError('Missing OAuth code.')
        text = codes[0]
    if not text or len(text) > 4096 or any(char.isspace() for char in text):
        raise ValueError('Invalid OAuth code.')
    return text
 
 
def _exchange(code: str, verifier: str) -> str:
    """Use the fixed HTTPS endpoint with no proxy inheritance or redirects."""
    endpoint = TOKEN_ENDPOINTS['openrouter']
    parts = urlsplit(endpoint)
    if parts.scheme != 'https' or parts.hostname != 'openrouter.ai' or parts.username or parts.password:
        raise ValueError('Invalid OAuth token endpoint.')
    try:
        with httpx.Client(trust_env=False, follow_redirects=False, timeout=30.0) as client:
            response = client.post(endpoint, json={
                'code': code,
                'code_verifier': verifier,
                'code_challenge_method': 'S256',
            })
            if not 200 <= response.status_code < 300:
                raise ValueError('OAuth exchange refused.')
            payload = response.json()
    except Exception:
        raise ValueError('OAuth exchange failed; run /login openrouter again.') from None
    value = payload.get('key') if isinstance(payload, dict) else None
    if not isinstance(value, str) or not value.strip():
        raise ValueError('OAuth exchange returned no usable key.')
    return value
 
 
class OpenRouterOAuth:
    """Optional capability for the OpenRouter catalogue row only."""
 
    def oauth_login(self, callbacks: OAuthLoginCallbacks) -> OAuthCredentials:
        """Display the URL and exchange the pasted code without opening a UI."""
        verifier = secrets.token_urlsafe(32)
        challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode('ascii')).digest()).rstrip(b'=').decode('ascii')
        url = AUTHORIZE_URL + '?' + urlencode({
            'callback_url': CALLBACK_URL,
            'code_challenge': challenge,
            'code_challenge_method': 'S256',
        })
        callbacks.on_auth(url)
        pasted = callbacks.on_prompt('Paste the code or final redirect URL from the browser: ')
        code = _pasted_code(pasted)
        return OAuthCredentials(access=_exchange(code, verifier), obtained_at=time.time())
 
    def oauth_refresh(self, creds: OAuthCredentials) -> OAuthCredentials:
        """OpenRouter keys do not require a refresh flow in this first lane."""
        raise NotImplementedError('Refresh is unavailable; run /login openrouter.')
 
    def oauth_api_key(self, creds: OAuthCredentials) -> str:
        """Return the bearer string expected by the existing provider client."""
        return creds.access