feat: build core/auth JWT validation and wire it into get_caller_context

# CALLBACK_SETTINGS__TIMEOUT=10.0

# QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c"

# LOCATION_RETRIEVAL_SETTINGS__SERVICE_SPECIFICATION_ID="0129e7ce-8e02-5dfd-bae1-0303bcc7676e"

# KEYCLOAK_SETTINGS__* default to a deliberately-unreachable `.invalid` host so a

# misconfigured deployment fails loudly instead of silently trusting the wrong

# Keycloak. Point these at your actual realm, e.g. (matches the oop-platform-chart):

# KEYCLOAK_SETTINGS__ISSUER="http://localhost:8080/realms/oop"

# KEYCLOAK_SETTINGS__JWKS_URL="http://localhost:8080/realms/oop/protocol/openid-connect/certs"

# KEYCLOAK_SETTINGS__AUDIENCE="oeg"

import asyncio

from typing import Any

import jwt

import structlog

from jwt import PyJWKClient

from jwt.exceptions import PyJWKClientError

from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.exceptions import DownstreamServiceException, UnauthorizedException

logger = structlog.get_logger(__name__)

class KeycloakTokenVerifier:

    def __init__(self) -> None:

        settings = get_settings().keycloak_settings

        self._issuer = settings.issuer

        self._audience = settings.audience

        self._algorithms = settings.algorithms

        self._jwks_client = PyJWKClient(str(settings.jwks_url))

    async def verify(self, token: str) -> dict[str, Any]:

        try:

            # get_signing_key_from_jwt fetches/caches Keycloak's JWKS over HTTP via

            # urllib, which is blocking -- push it to a thread so it doesn't stall

            # the event loop the way the rest of this codebase's httpx calls don't.

            signing_key = await asyncio.to_thread(self._jwks_client.get_signing_key_from_jwt, token)

        except PyJWKClientError as exc:

            logger.error("keycloak_jwks_fetch_failed", error=str(exc))

            raise DownstreamServiceException(message="Could not reach Keycloak") from exc

        try:

            return jwt.decode(

                token,

                signing_key.key,

                algorithms=self._algorithms,

                audience=self._audience,

                issuer=self._issuer,

            )

        except jwt.PyJWTError as exc:

            logger.warning("invalid_access_token", error=str(exc))

            raise UnauthorizedException(message="Invalid access token") from exc

import asyncio

from typing import Any

from jwt import PyJWKClient

from jwt.exceptions import PyJWKClientError

class JWKSCache:

    """Wraps PyJWT's PyJWKClient, which already caches signing keys by `kid` and

    refreshes on a cache miss (key rotation) or TTL expiry -- so this doesn't fetch

    per request (architecture.md E3). Also tracks last-fetch health for /readyz.

    """

    def __init__(self, jwks_url: str, lifespan: float = 300) -> None:

        self._client = PyJWKClient(jwks_url, cache_keys=True, lifespan=lifespan)

        self._healthy = False

    @property

    def is_healthy(self) -> bool:

        return self._healthy

    async def warm(self) -> None:

        """Fetch the JWK set once at startup instead of waiting for the first

        request, so /readyz reflects real JWKS health immediately."""

        try:

            await asyncio.to_thread(self._client.get_jwk_set)

        except PyJWKClientError:

            self._healthy = False

            return

        self._healthy = True

    async def get_signing_key(self, token: str) -> Any:

        try:

            key = await asyncio.to_thread(self._client.get_signing_key_from_jwt, token)

        except PyJWKClientError:

            self._healthy = False

            raise

        self._healthy = True

        return key

from dataclasses import dataclass

from enum import StrEnum

from functools import cached_property

from typing import Any

import jwt

import structlog

from jwt.exceptions import PyJWTError

from open_exposure_gateway.core.auth.jwks_cache import JWKSCache

from open_exposure_gateway.core.config import KeycloakSettings

from open_exposure_gateway.core.exceptions import UnauthorizedException

logger = structlog.get_logger(__name__)

# Keycloak signs with RS256; pinning this explicitly (rather than trusting the

# token header) is what prevents an algorithm-confusion attack.

_ALGORITHMS = ["RS256"]

class TokenFlow(StrEnum):

    TWO_LEGGED = "TWO_LEGGED"

    THREE_LEGGED = "THREE_LEGGED"

@dataclass

class DecodedClaims:

    claims: dict[str, Any]

    @cached_property

    def flow(self) -> TokenFlow:

        # auth_time is stamped onto a token only when its subject went through an

        # interactive login; a Client Credentials token has no user to authenticate

        # and never carries it. grant_type itself isn't in the token, so this is

        # the one reliable signal for which flow issued it.

        return TokenFlow.THREE_LEGGED if "auth_time" in self.claims else TokenFlow.TWO_LEGGED

async def validate_token(

    token: str, jwks_cache: JWKSCache, settings: KeycloakSettings

) -> DecodedClaims:

    try:

        signing_key = await jwks_cache.get_signing_key(token)

        claims = jwt.decode(

            token,

            signing_key.key,

            algorithms=_ALGORITHMS,

            issuer=settings.issuer,

            audience=settings.audience,

        )

    except PyJWTError as exc:

        logger.warning("invalid_access_token", error=str(exc))

        raise UnauthorizedException(message="Invalid access token") from exc

    return DecodedClaims(claims=claims)