provider_oauth.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/provider_oauth.py
 
"""Provider OAuth credentials, stored separately from the flat secrets map.
 
Purpose
-------
Keep provider authentication records private and resolve the runtime credential.
 
Responsibilities
----------------
Strict reads, atomic writes, isolated deletion and shared credential resolution.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - corrupt store notification without contents
    L3 - lifecycle counts
    L4 - module lifecycle
 
Contracts
---------
No credential reaches print, logging, diagnostics or exception messages.
Reads are strict; runtime resolution degrades on a malformed sidecar.
Expiry refuses use and never triggers an automatic refresh.
 
Dependencies
------------
- oc_diagnostics (structured diagnostics)
- option_c_agent_contracts.provider_plugin (OAuth value types and the provider plugin interface)
- oc_chat.core.provider_secrets (gitignore and atomic mode helpers)
"""
from __future__ import annotations
 
import json
import math
import re
import time
from collections.abc import Mapping
from dataclasses import asdict
from pathlib import Path
 
from oc_diagnostics import _dbg
from option_c_agent_contracts.provider_plugin import OAuthCredentials, ProviderPlugin
from oc_chat.core.provider_secrets import (
    SecretsFileError, _chmod_secret, _write_atomically, assert_gitignored,
)
 
_dbg('OC_CHAT.MAIN', 'module', 'loaded', 4)
 
DEFAULT_OAUTH_PATH = Path('.option_c/provider_oauth.json')
_PROVIDER_NAME = re.compile(r'[a-z][a-z0-9_]*\Z')
 
 
def _target(path: str | Path | None) -> Path:
    return DEFAULT_OAUTH_PATH if path is None else Path(path)
 
 
def _valid_record(record: object) -> bool:
    if not isinstance(record, dict):
        return False
    if not {'access', 'obtained_at'} <= record.keys():
        return False
    if record.keys() - {'access', 'refresh', 'expires_at', 'obtained_at'}:
        return False
    if not isinstance(record['access'], str) or not record['access'].strip():
        return False
    refresh = record.get('refresh')
    if refresh is not None and (not isinstance(refresh, str) or not refresh.strip()):
        return False
    for name in ('obtained_at', 'expires_at'):
        value = record.get(name)
        if value is None and name == 'expires_at':
            continue
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            return False
        try:
            if not math.isfinite(value) or value < 0:
                return False
        except OverflowError:
            return False
    return True
 
 
def load_credentials(path: str | Path | None = None) -> dict[str, OAuthCredentials]:
    """Read strictly, returning an empty store only when the file is absent."""
    source = _target(path)
    try:
        text = source.read_text(encoding='utf-8')
    except FileNotFoundError:
        return {}
    except (OSError, UnicodeError):
        raise SecretsFileError('Could not read the provider OAuth sidecar.') from None
    try:
        payload = json.loads(text)
    except (ValueError, RecursionError):
        raise SecretsFileError('Could not parse the provider OAuth sidecar.') from None
    if not isinstance(payload, dict) or any(
        not _PROVIDER_NAME.fullmatch(name) or not _valid_record(record)
        for name, record in payload.items()
    ):
        raise SecretsFileError('Invalid provider OAuth sidecar shape.')
    return {name: OAuthCredentials(**record) for name, record in payload.items()}
 
 
def write_credentials(
    provider_name: str,
    credentials: OAuthCredentials | None,
    *,
    path: str | Path | None = None,
) -> Path:
    """Replace one provider record, or remove only that record on logout."""
    if not _PROVIDER_NAME.fullmatch(provider_name):
        raise ValueError('Invalid OAuth provider name.')
    if credentials is not None and not _valid_record(asdict(credentials)):
        raise ValueError('Invalid OAuth credential record.')
    target = assert_gitignored(_target(path))
    payload = {name: asdict(creds) for name, creds in load_credentials(target).items()}
    if credentials is None:
        payload.pop(provider_name, None)
    else:
        payload[provider_name] = asdict(credentials)
    _write_atomically(payload, target)
    _chmod_secret(target)
    return target
 
 
def resolve_api_key(
    provider: ProviderPlugin,
    api_key: str | None = None,
    *,
    environ: Mapping[str, str] | None = None,
) -> str:
    """Resolve explicit argument, environment, then a usable stored credential."""
    import os
    func = "resolve_api_key"
    if api_key is not None:
        return api_key
    env = os.environ if environ is None else environ
    value = env.get(provider.api_key_env_var, '') if provider.api_key_env_var else ''
    if value.strip():
        return value
    capability = getattr(provider, 'oauth', None)
    if capability is None:
        return ''
    try:
        credentials = load_credentials().get(provider.name)
    except SecretsFileError:
        _dbg('OC_CHAT.MAIN', func, 'Provider OAuth sidecar could not be loaded.', 2)
        return ''
    if credentials is None:
        return ''
    message = f'Credential unavailable for {provider.name}; run /login {provider.name}.'
    if credentials.expires_at is not None and credentials.expires_at <= time.time():
        raise ValueError(message)
    try:
        value = capability.oauth_api_key(credentials)
    except Exception:
        raise ValueError(message) from None
    if not isinstance(value, str) or not value.strip():
        raise ValueError(message)
    return value
 
 
def has_usable_credentials(provider: ProviderPlugin) -> bool:
    """Presence-only gate; invalid or expired credentials never pass."""
    try:
        return bool(resolve_api_key(provider, environ={}))
    except ValueError:
        return False