mcphub/core/oauth/client_registry.py

"""
OAuth 2.1 Client Registry
Manages OAuth client registration and validation
"""

import hashlib
import json
import logging
import secrets
from pathlib import Path

from .schemas import OAuthClient

logger = logging.getLogger(__name__)

# Default data directory: /app/data in Docker, ./data elsewhere
_DEFAULT_DATA_DIR = "/app/data" if Path("/app").exists() else "./data"


class ClientRegistry:
    """
    OAuth Client Registry with JSON storage.

    Storage: data/oauth_clients.json
    """

    def __init__(self, data_dir: str | None = None):
        data_dir = data_dir or _DEFAULT_DATA_DIR
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)

        self.clients_file = self.data_dir / "oauth_clients.json"

        # Initialize if not exists
        if not self.clients_file.exists():
            self._write_clients({})

        # Load clients into memory (small dataset)
        self.clients: dict[str, OAuthClient] = self._load_clients()

    def _read_clients(self) -> dict:
        """Read clients JSON file"""
        try:
            with open(self.clients_file) as f:
                return json.load(f)
        except Exception as e:
            logger.error(f"Error reading clients file: {e}")
            return {}

    def _write_clients(self, data: dict) -> bool:
        """Write clients JSON file"""
        try:
            with open(self.clients_file, "w") as f:
                json.dump(data, f, indent=2, default=str)
            return True
        except Exception as e:
            logger.error(f"Error writing clients file: {e}")
            return False

    def _load_clients(self) -> dict[str, OAuthClient]:
        """Load clients from file"""
        data = self._read_clients()
        return {client_id: OAuthClient(**client_data) for client_id, client_data in data.items()}

    def _save_clients(self) -> bool:
        """Save clients to file"""
        data = {
            client_id: client.model_dump(mode="json") for client_id, client in self.clients.items()
        }
        return self._write_clients(data)

    def create_client(
        self,
        client_name: str,
        redirect_uris: list[str],
        grant_types: list[str] | None = None,
        allowed_scopes: list[str] | None = None,
        metadata: dict | None = None,
    ) -> tuple[str, str]:
        """
        Create new OAuth client.

        Returns:
            (client_id, client_secret) tuple
        """
        # Generate client_id and client_secret
        client_id = f"cmp_client_{secrets.token_urlsafe(16)}"
        client_secret = secrets.token_urlsafe(32)

        # Hash client secret
        client_secret_hash = hashlib.sha256(client_secret.encode()).hexdigest()

        # Create client
        client = OAuthClient(
            client_id=client_id,
            client_secret_hash=client_secret_hash,
            client_name=client_name,
            redirect_uris=redirect_uris,
            grant_types=grant_types or ["authorization_code", "refresh_token"],
            allowed_scopes=allowed_scopes or ["read", "write"],
            metadata=metadata or {},
        )

        # Save
        self.clients[client_id] = client
        self._save_clients()

        logger.info(f"Created OAuth client: {client_id} ({client_name})")

        # Return client_secret in plain text (only time it's visible)
        return client_id, client_secret

    def get_client(self, client_id: str) -> OAuthClient | None:
        """Get client by ID"""
        return self.clients.get(client_id)

    def list_clients(self) -> list[OAuthClient]:
        """List all clients"""
        return list(self.clients.values())

    def validate_client_secret(self, client_id: str, client_secret: str) -> bool:
        """
        Validate client secret.

        Args:
            client_id: Client ID
            client_secret: Plain text client secret

        Returns:
            True if valid, False otherwise
        """
        client = self.get_client(client_id)
        if not client:
            return False

        # Hash provided secret
        secret_hash = hashlib.sha256(client_secret.encode()).hexdigest()

        # Constant-time comparison
        return secrets.compare_digest(secret_hash, client.client_secret_hash)

    def delete_client(self, client_id: str) -> bool:
        """Delete OAuth client"""
        if client_id in self.clients:
            del self.clients[client_id]
            self._save_clients()
            logger.info(f"Deleted OAuth client: {client_id}")
            return True
        return False


# Singleton instance
_client_registry: ClientRegistry | None = None


def get_client_registry() -> ClientRegistry:
    """Get singleton ClientRegistry instance"""
    global _client_registry
    if _client_registry is None:
        _client_registry = ClientRegistry()
    return _client_registry