Repository: apache/airflow
File: providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py (Python)

Code before:
            data=self._get_batch_payload(client_id, permissions),
            headers=self._get_headers(user.access_token),
            timeout=5,
        )

        if resp.status_code == 200:
            return {(perm["scopes"][0], perm["rsname"]) for perm in resp.json()}
        if resp.status_code == 401:
            log.debug("Received 401 from Keycloak: %s", resp.text)
            return set()
        if resp.status_code == 403:
            return set()
        if resp.status_code == 400:
            error = json.loads(resp.text)
            raise AirflowException(
                f"Request not recognized by Keycloak. {error.get('error')}. {error.get('error_description')}"
            )
        raise AirflowException(f"Unexpected error: {resp.status_code} - {resp.text}")

    @staticmethod
    def _get_token_url(server_url, realm):
        # Normalize server_url to avoid double slashes (required for Keycloak 26.4+ strict path validation).
        return f"{server_url.rstrip('/')}/realms/{realm}/protocol/openid-connect/token"

    @staticmethod
    def _get_team_name(details: Any | None) -> str | None:
        return getattr(details, "team_name", None) if details else None

    @staticmethod
    def _get_payload(client_id: str, permission: str, attributes: dict[str, str] | None = None):
        payload: dict[str, Any] = {
            "grant_type": "urn:ietf:params:oauth:grant-type:uma-ticket",
            "audience": client_id,
            "permission": permission,
        }
        if attributes:
            # Per UMA spec, push claims using claim_token parameter with base64-encoded JSON
            # Values must be arrays of strings per Keycloak documentation
            # See: https://www.keycloak.org/docs/latest/authorization_services/index.html#_service_pushing_claims
            claims = {key: [value] for key, value in attributes.items()}
            claim_json = json.dumps(claims, sort_keys=True)
            claim_token = base64.b64encode(claim_json.encode()).decode()
            payload["claim_token"] = claim_token
            payload["claim_token_format"] = "urn:ietf:params:oauth:token-type:jwt"

        return payload

    @staticmethod
    def _get_batch_payload(client_id: str, permissions: list[tuple[ExtendedResourceMethod, str]]):
        payload: dict[str, Any] = {
            "grant_type": "urn:ietf:params:oauth:grant-type:uma-ticket",

Review comment (suggestion):
Nit: instead of using `Any` maybe you can list the types like `ConnectionsDetails | VariableDetails | ...`

Code after:
        server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)

        resp = self.http_session.post(
            self._get_token_url(server_url, realm),
            data=self._get_batch_payload(client_id, permissions),
            headers=self._get_headers(user.access_token),
            timeout=5,
        )

        if resp.status_code == 200:
            return {(perm["scopes"][0], perm["rsname"]) for perm in resp.json()}
        if resp.status_code == 401:
            log.debug("Received 401 from Keycloak: %s", resp.text)
            return set()
        if resp.status_code == 403:
            return set()
        if resp.status_code == 400:
            error = json.loads(resp.text)
            raise AirflowException(
                f"Request not recognized by Keycloak. {error.get('error')}. {error.get('error_description')}"
            )
        raise AirflowException(f"Unexpected error: {resp.status_code} - {resp.text}")

    @staticmethod
    def _get_token_url(server_url, realm):
        # Normalize server_url to avoid double slashes (required for Keycloak 26.4+ strict path validation).
        return f"{server_url.rstrip('/')}/realms/{realm}/protocol/openid-connect/token"

    @staticmethod
    def _get_team_name(
        details: ConnectionDetails | DagDetails | PoolDetails | VariableDetails | None,
    ) -> str | None:
        return getattr(details, "team_name", None) if details else None

    @staticmethod
    def _get_payload(client_id: str, permission: str, attributes: dict[str, str] | None = None):
        payload: dict[str, Any] = {
            "grant_type": "urn:ietf:params:oauth:grant-type:uma-ticket",
            "audience": client_id,
            "permission": permission,
        }
        if attributes:
            # Per UMA spec, push claims using claim_token parameter with base64-encoded JSON
            # Values must be arrays of strings per Keycloak documentation
            # See: https://www.keycloak.org/docs/latest/authorization_services/index.html#_service_pushing_claims
            claims = {key: [value] for key, value in attributes.items()}
            claim_json = json.dumps(claims, sort_keys=True)
            claim_token = base64.b64encode(claim_json.encode()).decode()
            payload["claim_token"] = claim_token
            payload["claim_token_format"] = "urn:ietf:params:oauth:token-type:jwt"