Class OidcTokenParser

sdk/python/feast/permissions/auth/oidc_token_parser.py:23–265

A ``TokenParser`` to use an OIDC server to retrieve the user details. Server settings are retrieved from the ``auth`` configuration of the Feature store. Incoming tokens that contain a ``kubernetes.io`` claim (i.e. Kubernetes service-account tokens) are handled via a lightweight To

class OidcTokenParser(TokenParser):
    """
    A ``TokenParser`` to use an OIDC server to retrieve the user details.
    Server settings are retrieved from the ``auth`` configuration of the Feature store.

    Incoming tokens that contain a ``kubernetes.io`` claim (i.e. Kubernetes
    service-account tokens) are handled via a lightweight TokenReview that
    extracts only the namespace — no RBAC queries needed.  All other tokens
    follow the standard OIDC/Keycloak JWKS validation path.
    """

    _auth_config: OidcAuthConfig

    def __init__(self, auth_config: OidcAuthConfig):
        self._auth_config = auth_config
        self.oidc_discovery_service = OIDCDiscoveryService(
            self._auth_config.auth_discovery_url,
            verify_ssl=self._auth_config.verify_ssl,
            ca_cert_path=self._auth_config.ca_cert_path,
        )
        self._k8s_auth_api = None

    async def _validate_token(self, access_token: str):
        """
        Validate the token extracted from the header of the user request against the OAuth2 server.
        """
        # FastAPI's OAuth2AuthorizationCodeBearer requires a Request type but actually uses only the headers field
        # https://github.com/tiangolo/fastapi/blob/eca465f4c96acc5f6a22e92fd2211675ca8a20c8/fastapi/security/oauth2.py#L380
        request = Mock(spec=Request)
        request.headers = {"Authorization": f"Bearer {access_token}"}

        oauth_2_scheme = OAuth2AuthorizationCodeBearer(
            tokenUrl=self.oidc_discovery_service.get_token_url(),
            authorizationUrl=self.oidc_discovery_service.get_authorization_url(),
            refreshUrl=self.oidc_discovery_service.get_refresh_url(),
        )

        await oauth_2_scheme(request=request)

    @staticmethod
    def _extract_username_or_raise_error(data: dict) -> str:
        """Extract the username from the decoded JWT. Raises if missing — identity is mandatory.

        Checks ``preferred_username`` first (Keycloak default), then falls back
        to ``upn`` (Azure AD / Entra ID).
        """
        if "preferred_username" in data:
            return data["preferred_username"]
        if "upn" in data:
            return data["upn"]
        raise AuthenticationError(
            "Missing preferred_username or upn field in access token."
        )

    @staticmethod
    def _extract_claim(data: dict, *keys: str, expected_type: type = list):
        """Walk *keys* into *data* and return the leaf value, or ``expected_type()`` if any key is missing or the wrong type."""
        node = data