feat: implement Keycloak OIDC client with token handling and error management

"""

Keycloak's OpenID Connect endpoints, over httpx.

Only the authorization code flow with PKCE (RFC 7636) is used: the realm has the

password grant disabled, and the client is confidential, so the code exchange also

carries the client secret. Tokens are never logged.

"""

from typing import Any

from urllib.parse import urlencode

import httpx

import structlog

from pydantic import BaseModel, ValidationError

from portal_service.config import OIDCSettings

from portal_service.domain.errors import (

    IdentityProviderUnavailableError,

    InvalidGrantError,

)

from portal_service.domain.models import IssuedTokens, SessionTokens

from portal_service.domain.ports.identity_provider import IdentityProvider

logger: structlog.BoundLogger = structlog.get_logger(__name__)

_SCOPE = "openid profile email"

class _TokenResponse(BaseModel):

    """The fields of Keycloak's token response that we use."""

    access_token: str

    expires_in: int

    refresh_token: str | None = None

    refresh_expires_in: int | None = None

    id_token: str | None = None

class KeycloakOIDCClient(IdentityProvider):

    def __init__(

        self,

        settings: OIDCSettings,

        transport: httpx.AsyncBaseTransport | None = None,

    ) -> None:

        self._settings = settings

        # Tests pass an httpx.MockTransport here; in production it stays None.

        self._transport = transport

    def authorization_url(self, *, state: str, nonce: str, code_challenge: str) -> str:

        params = {

            "client_id": self._settings.client_id,

            "response_type": "code",

            "scope": _SCOPE,

            "redirect_uri": self._settings.redirect_uri,

            "state": state,

            "nonce": nonce,

            "code_challenge": code_challenge,

            "code_challenge_method": "S256",

        }

        return f"{self._settings.authorization_endpoint}?{urlencode(params)}"

    async def exchange_code(self, *, code: str, code_verifier: str) -> IssuedTokens:

        return await self._token_request(

            {

                "grant_type": "authorization_code",

                "code": code,

                "redirect_uri": self._settings.redirect_uri,

                "code_verifier": code_verifier,

            }

        )

    async def refresh(self, refresh_token: str) -> IssuedTokens:

        return await self._token_request(

            {"grant_type": "refresh_token", "refresh_token": refresh_token}

        )

    def logout_url(self, *, id_token: str, post_logout_redirect_uri: str) -> str:

        params = {

            "id_token_hint": id_token,

            "post_logout_redirect_uri": post_logout_redirect_uri,

            "client_id": self._settings.client_id,

        }

        return f"{self._settings.end_session_endpoint}?{urlencode(params)}"

    async def _token_request(self, form: dict[str, str]) -> IssuedTokens:

        data = {

            **form,

            "client_id": self._settings.client_id,

            "client_secret": self._settings.client_secret,

        }

        try:

            async with httpx.AsyncClient(

                timeout=self._settings.timeout, transport=self._transport

            ) as client:

                response = await client.post(

                    self._settings.token_endpoint,

                    data=data,

                    headers={"Accept": "application/json"},

                )

        except httpx.HTTPError as exc:

            logger.error("token endpoint unreachable", grant=form["grant_type"], error=str(exc))

            raise IdentityProviderUnavailableError("Token endpoint unreachable") from exc

        if response.status_code >= 400:

            self._raise_for_error(response, form["grant_type"])

        return self._to_issued_tokens(response)

    def _raise_for_error(self, response: httpx.Response, grant: str) -> None:

        error = _oauth_error(response)

        logger.warning(

            "token request rejected",

            grant=grant,

            status=response.status_code,

            oauth_error=error,

        )

        if response.status_code == 400:

            # RFC 6749 §5.2: a bad, expired or already-used code or refresh token.

            raise InvalidGrantError(f"Token request rejected: {error or 'invalid_request'}")

        # 401 and 403 mean our own client credentials are wrong, 5xx means Keycloak

        # is unwell. Neither is the user's fault, so neither is an AuthenticationError.

        raise IdentityProviderUnavailableError(

            f"Token endpoint returned {response.status_code}: {error or 'no error code'}"

        )

    def _to_issued_tokens(self, response: httpx.Response) -> IssuedTokens:

        try:

            parsed = _TokenResponse.model_validate_json(response.content)

        except ValidationError as exc:

            # Field names only: the payload holds tokens and must not reach the logs.

            fields = ", ".join(str(err["loc"][0]) for err in exc.errors() if err["loc"])

            logger.error("token response not usable", fields=fields)

            raise IdentityProviderUnavailableError(f"Token response invalid: {fields}") from exc

        return IssuedTokens(

            tokens=SessionTokens(

                access_token=parsed.access_token,

                refresh_token=parsed.refresh_token,

                id_token=parsed.id_token,

            ),

            expires_in=parsed.expires_in,

            refresh_expires_in=parsed.refresh_expires_in,

        )

def _oauth_error(response: httpx.Response) -> str | None:

    """The `error` field of an OAuth error response, if the body has one."""

    try:

        body: Any = response.json()

    except ValueError:

        return None

    return body.get("error") if isinstance(body, dict) else None

    pass

# Keycloak refused a login code or refresh token: expired, already used, or not ours.

class InvalidGrantError(AuthenticationError):

    pass

# Not an AuthenticationError: the token was never judged, so it is not the caller's fault.

class IdentityProviderUnavailableError(Exception):

    pass

    id_token: str | None = Field(default=None, repr=False)

class IssuedTokens(BaseModel):

    """

    One answer from Keycloak's token endpoint, for a login or a refresh.

    `expires_in` is how many seconds the access token stays valid.

    """

    tokens: SessionTokens

    expires_in: int

    refresh_expires_in: int | None = None

class LoginSession(BaseModel):

    """

    A stored login session. `id` is the SHA-256 of the session id in the cookie, and

from portal_service.domain.ports.identity_provider import IdentityProvider

__all__ = [

    "IdentityProvider",

]