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

Code before:
        self,
        *,
        method: ResourceMethod | str,
        resource_type: KeycloakResource,
        user: KeycloakAuthManagerUser,
        resource_id: str | None = None,
        team_name: str | None = None,
        attributes: dict[str, str | None] | None = None,
    ) -> bool:
        client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
        realm = conf.get(CONF_SECTION_NAME, CONF_REALM_KEY)
        server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)

        context_attributes = prune_dict(attributes or {})
        if resource_id:
            context_attributes[RESOURCE_ID_ATTRIBUTE_NAME] = resource_id
        elif method == "GET":
            method = "LIST"

        is_multi_team = conf.getboolean("core", "multi_team", fallback=False)
        is_team_scoped = resource_type in TEAM_SCOPED_RESOURCES
        is_teamless = team_name is None

        # Team-scoped resources require a team, except for LIST which uses global permission.
        if is_multi_team and is_team_scoped and is_teamless and method != "LIST":
            raise ValueError("Missing team_name for team-scoped resource in multi-team mode.")
        if is_multi_team and is_team_scoped and is_teamless and method == "LIST":
            permission = f"{resource_type.value}#{method}"
        else:
            resource_name = self._get_resource_name(resource_type, team_name)
            if resource_name is None:
                return False
            permission = f"{resource_name}#{method}"

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

        if resp.status_code == 200:
            return True
        if resp.status_code == 401:
            log.debug("Received 401 from Keycloak: %s", resp.text)
            return False
        if resp.status_code == 403:
            return False
        if resp.status_code == 400:
            error = json.loads(resp.text)
            raise AirflowException(

Review comment (suggestion):
We should not do that. In a multi-team environment, Dags, Connections, Variables, Pools can also be global. They must not belong to a team

Code after:
        self,
        *,
        method: ResourceMethod | str,
        resource_type: KeycloakResource,
        user: KeycloakAuthManagerUser,
        resource_id: str | None = None,
        team_name: str | None = None,
        attributes: dict[str, str | None] | None = None,
    ) -> bool:
        client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
        realm = conf.get(CONF_SECTION_NAME, CONF_REALM_KEY)
        server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)

        context_attributes = prune_dict(attributes or {})
        if resource_id:
            context_attributes[RESOURCE_ID_ATTRIBUTE_NAME] = resource_id
        elif method == "GET":
            method = "LIST"

        if (
            team_name
            and conf.getboolean("core", "multi_team", fallback=False)
            and resource_type in TEAM_SCOPED_RESOURCES
        ):
            resource_name = f"{resource_type.value}:{team_name}"
        else:
            resource_name = resource_type.value
        permission = f"{resource_name}#{method}"

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

        if resp.status_code == 200:
            return True
        if resp.status_code == 401:
            log.debug("Received 401 from Keycloak: %s", resp.text)
            return False
        if resp.status_code == 403:
            return False
        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}")

    def _is_batch_authorized(