Newer
Older
navi-1 / navi / auth / client.py
"""GAuthClient singleton wired to navi config.

Supports dynamic redirect_uri for multi-domain deployments while keeping
state and PKCE stores shared across all client instances.
"""

from gnexus_gauth.client import GAuthClient
from gnexus_gauth.config import GAuthConfig
from gnexus_gauth.oauth import HttpTokenEndpoint
from gnexus_gauth.runtime import HttpRuntimeUserProvider
from gnexus_gauth.support import InMemoryPkceStore, InMemoryStateStore
from gnexus_gauth.webhook import HmacWebhookVerifier, JsonWebhookParser

from navi.config import settings

# Shared ephemeral stores — must be singletons so state/PKCE survive across
# per-request client instances with different redirect_uris.
_state_store: InMemoryStateStore | None = None
_pkce_store: InMemoryPkceStore | None = None

# Default client (uses redirect_uri from settings) — used for token refresh,
# fetch_user, and webhook parsing where redirect_uri is irrelevant.
_default_gauth_client: GAuthClient | None = None


def _ensure_stores() -> tuple[InMemoryStateStore, InMemoryPkceStore]:
    """Create shared stores if they don't exist."""
    global _state_store, _pkce_store
    if _state_store is None:
        _state_store = InMemoryStateStore()
    if _pkce_store is None:
        _pkce_store = InMemoryPkceStore()
    return _state_store, _pkce_store


def _make_client(redirect_uri: str) -> GAuthClient:
    """Build a GAuthClient with the given redirect_uri and shared stores."""
    state_store, pkce_store = _ensure_stores()
    config = GAuthConfig(
        base_url=settings.gnauth_base_url,
        client_id=settings.gnauth_client_id,
        client_secret=settings.gnauth_client_secret,
        redirect_uri=redirect_uri,
    )
    return GAuthClient(
        config=config,
        token_endpoint=HttpTokenEndpoint(config),
        runtime_user_provider=HttpRuntimeUserProvider(config),
        webhook_verifier=HmacWebhookVerifier(config),
        webhook_parser=JsonWebhookParser(),
        state_store=state_store,
        pkce_store=pkce_store,
    )


def get_gauth_client(redirect_uri: str | None = None) -> GAuthClient:
    """Return a GAuthClient.

    When *redirect_uri* is provided (e.g. OAuth login/callback), a fresh
    client is built with that URI while keeping the shared state/PKCE stores.

    When *redirect_uri* is omitted, the cached default client (using the
    redirect_uri from settings) is returned.  This default is safe for
    token refresh, fetch_user, and webhook parsing where redirect_uri is
    not used.
    """
    global _default_gauth_client
    if redirect_uri is not None:
        return _make_client(redirect_uri)
    if _default_gauth_client is None:
        _default_gauth_client = _make_client(settings.gnauth_redirect_uri)
    return _default_gauth_client